Today’s focus is on implementing transfer learning using ResNet50 to classify images of dogs and wolves. We’ll walk through data preparation, dataset loading with augmentation, model fine-tuning, training, and prediction visualization—all within MindSpore.

Data Preparation

We begin by downloading a curated dataset of dog and wolf images sourced from ImageNet. Each class contains approximately 120 training and 30 validation images. The dataset is downloaded and automatically extracted using the following code: ``` from download import download

dataset_url = "https://mindspore-website.obs.cn-north-4.myhuaweicloud.com/notebook/datasets/intermediate/Canidae_data.zip" download(dataset_url, "./datasets-Canidae", kind="zip", replace=True)

The extracted structure contains two subdirectories: `train/` and `val/`, each organized by class label. ### Loading and Augmenting the Dataset

We use `mindspore.dataset.ImageFolderDataset` to load the images and apply transformations based on training or validation mode. Key hyperparameters are defined as follows: ```
batch_size = 18
image_size = 224
num_epochs = 5
lr = 0.001
momentum = 0.9
workers = 4

The data loader function applies different augmentations for training and validation: - Training: Random crop, horizontal flip, normalization, and channel rearrangement.

• Validation: Resize, center crop, normalization, and channel rearrangement (no randomness).

import mindspore as ms
import mindspore.dataset as ds
import mindspore.dataset.vision as vision

data_path_train = "./datasets-Canidae/data/Canidae/train/"
data_path_val = "./datasets-Canidae/data/Canidae/val/"

mean = [0.485 * 255, 0.456 * 255, 0.406 * 255]
std = [0.229 * 255, 0.224 * 255, 0.225 * 255]
scale = 32

def create_dataset_canidae(dataset_path, usage):
    data_set = ds.ImageFolderDataset(dataset_path, num_parallel_workers=workers, shuffle=True)

    if usage == "train":
        trans = [
            vision.RandomCropDecodeResize(size=image_size, scale=(0.08, 1.0), ratio=(0.75, 1.333)),
            vision.RandomHorizontalFlip(prob=0.5),
            vision.Normalize(mean=mean, std=std),
            vision.HWC2CHW()
        ]
    else:
        trans = [
            vision.Decode(),
            vision.Resize(image_size + scale),
            vision.CenterCrop(image_size),
            vision.Normalize(mean=mean, std=std),
            vision.HWC2CHW()
        ]

    data_set = data_set.map(operations=trans, input_columns='image', num_parallel_workers=workers)
    data_set = data_set.batch(batch_size)
    return data_set

dataset_train = create_dataset_canidae(data_path_train, "train")
dataset_val = create_dataset_canidae(data_path_val, "val")

We visualize a batch of training samples to inspect the augmented images and their labels: ``` import matplotlib.pyplot as plt import numpy as np

class_name = {0: "dogs", 1: "wolves"}

data = next(dataset_train.create_dict_iterator()) images = data["image"].asnumpy() labels = data["label"].asnumpy()

plt.figure(figsize=(5, 5)) for i in range(4): img = images[i] img = np.transpose(img, (1, 2, 0)) img = img * std[:3] + mean[:3] # Denormalize img = np.clip(img, 0, 1)

plt.subplot(2, 2, i+1)
plt.imshow(img)
plt.title(class_name[int(labels[i])], fontsize=10)
plt.axis("off")

plt.show()

### Building and Loading ResNet50 with Pretrained Weights

We implement a modular ResNet50 architecture with residual blocks. The model consists of convolutional layers, batch normalization, ReLU activations, and skip connections. The final fully connected layer is replaced to match our two-class classification task. ```
from mindspore import nn, load_checkpoint, load_param_into_net
from typing import Type, List, Optional, Union

weight_init = nn.Normal(sigma=0.02)
gamma_init = nn.Normal(sigma=0.02)

class ResidualBlock(nn.Cell):
    expansion = 4

    def __init__(self, in_channel: int, out_channel: int, stride: int = 1, down_sample: Optional[nn.Cell] = None):
        super(ResidualBlock, self).__init__()
        self.conv1 = nn.Conv2d(in_channel, out_channel, kernel_size=1, weight_init=weight_init)
        self.bn1 = nn.BatchNorm2d(out_channel, gamma_init=gamma_init)
        self.conv2 = nn.Conv2d(out_channel, out_channel, kernel_size=3, stride=stride, weight_init=weight_init)
        self.bn2 = nn.BatchNorm2d(out_channel, gamma_init=gamma_init)
        self.conv3 = nn.Conv2d(out_channel, out_channel * self.expansion, kernel_size=1, weight_init=weight_init)
        self.bn3 = nn.BatchNorm2d(out_channel * self.expansion, gamma_init=gamma_init)
        self.relu = nn.ReLU()
        self.down_sample = down_sample

    def construct(self, x):
        identity = x
        out = self.relu(self.bn1(self.conv1(x)))
        out = self.relu(self.bn2(self.conv2(out)))
        out = self.bn3(self.conv3(out))

        if self.down_sample is not None:
            identity = self.down_sample(x)
        out += identity
        return self.relu(out)

def make_layer(in_channels, block: Type[ResidualBlock], planes: int, blocks: int, stride: int = 1):
    down_sample = None
    if stride != 1 or in_channels != planes * block.expansion:
        down_sample = nn.SequentialCell([
            nn.Conv2d(in_channels, planes * block.expansion, kernel_size=1, stride=stride, weight_init=weight_init),
            nn.BatchNorm2d(planes * block.expansion, gamma_init=gamma_init)
        ])

    layers = [block(in_channels, planes, stride, down_sample)]
    for _ in range(1, blocks):
        layers.append(block(planes * block.expansion, planes))

    return nn.SequentialCell(layers)

class ResNet(nn.Cell):
    def __init__(self, block: Type[ResidualBlock], layers: List[int], num_classes: int, input_channel: int):
        super(ResNet, self).__init__()
        self.conv1 = nn.Conv2d(3, 64, kernel_size=7, stride=2, weight_init=weight_init)
        self.bn1 = nn.BatchNorm2d(64)
        self.relu = nn.ReLU()
        self.max_pool = nn.MaxPool2d(kernel_size=3, stride=2, pad_mode="same")

        self.layer1 = make_layer(64, block, 64, layers[0])
        self.layer2 = make_layer(64 * block.expansion, block, 128, layers[1], stride=2)
        self.layer3 = make_layer(128 * block.expansion, block, 256, layers[2], stride=2)
        self.layer4 = make_layer(256 * block.expansion, block, 512, layers[3], stride=2)

        self.avg_pool = nn.AvgPool2d()
        self.flatten = nn.Flatten()
        self.fc = nn.Dense(input_channel, num_classes)

    def construct(self, x):
        x = self.relu(self.bn1(self.conv1(x)))
        x = self.max_pool(x)
        x = self.layer1(x)
        x = self.layer2(x)
        x = self.layer3(x)
        x = self.layer4(x)
        x = self.avg_pool(x)
        x = self.flatten(x)
        x = self.fc(x)
        return x

def resnet50(num_classes: int = 2, pretrained: bool = False):
    url = "https://mindspore-website.obs.cn-north-4.myhuaweicloud.com/notebook/models/application/resnet50_224_new.ckpt"
    ckpt_path = "./LoadPretrainedModel/resnet50_224_new.ckpt"

    if pretrained:
        download(url=url, path=ckpt_path, replace=True)
        model = ResNet(ResidualBlock, [3, 4, 6, 3], num_classes, 2048)
        param_dict = load_checkpoint(ckpt_path)
        load_param_into_net(model, param_dict)
        return model
    else:
        return ResNet(ResidualBlock, [3, 4, 6, 3], num_classes, 2048)

Freezing Features and Fine-Tuning

To leverage transfer learning, we freeze all layers except the final fully connected layer. This prevents gradient updates in the pretrained feature extractor and focuses training on adapting the classifier head: ``` import mindspore as ms from mindspore import train

net = resnet50(pretrained=True)

Replace the final layer for 2-class classification

in_features = net.fc.in_channels net.fc = nn.Dense(in_features, 2) net.avg_pool = nn.AvgPool2d(kernel_size=7) # Adjust for 224x224 input

Freeze all parameters except the final layer

for param in net.get_parameters(): if param.name not in ["fc.weight", "fc.bias"]: param.requires_grad = False

Define optimizer and loss

optimizer = nn.Momentum(params=net.trainable_params(), learning_rate=lr, momentum=momentum) loss_fn = nn.SoftmaxCrossEntropyWithLogits(sparse=True, reduction='mean')

def forward_fn(inputs, labels): logits = net(inputs) loss = loss_fn(logits, labels) return loss

grad_fn = ms.value_and_grad(forward_fn, None, optimizer.parameters)

def train_step(inputs, labels): loss, grads = grad_fn(inputs, labels) optimizer(grads) return loss

model = train.Model(net, loss_fn, optimizer, metrics={"Accuracy": train.Accuracy()})

### Training and Validation Loop

We train for 5 epochs, evaluating accuracy on the validation set after each epoch. The best-performing checkpoint is saved: ```
from mindspore import save_checkpoint
import os
import time

data_loader_train = dataset_train.create_tuple_iterator(num_epochs=num_epochs)
data_loader_val = dataset_val.create_tuple_iterator(num_epochs=num_epochs)

best_ckpt_dir = "./BestCheckpoint"
best_ckpt_path = os.path.join(best_ckpt_dir, "resnet50-best-freezing-param.ckpt")
best_acc = 0.0

if not os.path.exists(best_ckpt_dir):
    os.makedirs(best_ckpt_dir)

print("Starting training...")

for epoch in range(num_epochs):
    net.set_train()
    epoch_loss = []

    start_time = time.time()
    for images, labels in data_loader_train:
        labels = labels.astype(ms.int32)
        loss = train_step(images, labels)
        epoch_loss.append(loss.asnumpy())

    # Validate
    val_acc = model.eval(dataset_val)["Accuracy"]
    end_time = time.time()

    avg_loss = sum(epoch_loss) / len(epoch_loss)
    epoch_duration = (end_time - start_time) * 1000
    step_duration = epoch_duration / dataset_train.get_dataset_size()

    print(f"Epoch [{epoch+1}/{num_epochs}] "
          f"Loss: {avg_loss:.3f} | "
          f"Val Acc: {val_acc:.3f} | "
          f"Time: {epoch_duration:.1f}ms/epoch ({step_duration:.1f}ms/step)")

    if val_acc > best_acc:
        best_acc = val_acc
        save_checkpoint(net, best_ckpt_path)

print(f"\nTraining complete. Best validation accuracy: {best_acc:.3f}")

Visualizing Predictions on Validation Set

After training, we load the best checkpoint and visualize predictions on unseen validation images. Correct prediction are shown in blue; incorrect ones in red: ``` import numpy as np

def visualize_predictions(ckpt_path, val_dataset): net = resnet50(num_classes=2, pretrained=False) net.fc = nn.Dense(net.fc.in_channels, 2) net.avg_pool = nn.AvgPool2d(kernel_size=7)

param_dict = load_checkpoint(ckpt_path)
load_param_into_net(net, param_dict)
model = train.Model(net)

data = next(val_dataset.create_dict_iterator())
images = data["image"].asnumpy()
labels = data["label"].asnumpy()

outputs = model.predict(ms.Tensor(data["image"]))
preds = np.argmax(outputs.asnumpy(), axis=1)

plt.figure(figsize=(5, 5))
for i in range(4):
    img = images[i]
    img = np.transpose(img, (1, 2, 0))
    img = img * [0.229, 0.224, 0.225] + [0.485, 0.456, 0.406]
    img = np.clip(img, 0, 1)

    color = 'blue' if preds[i] == labels[i] else 'red'
    plt.subplot(2, 2, i+1)
    plt.imshow(img)
    plt.title(f"Pred: {class_name[preds[i]]}", color=color)
    plt.axis('off')

plt.tight_layout()
plt.show()

visualize_predictions(best_ckpt_path, dataset_val)