-
Notifications
You must be signed in to change notification settings - Fork 7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactoring and moving MobileNetV2 to make it reusable (#3177)
* Moving mobilenet.py to mobilenetv2.py * Adding mobilenet.py for BC. * Extending ConvBNReLU for reuse. * Reduce import scope on mobilenet to only the public and versioned classes and methods.
- Loading branch information
Showing
4 changed files
with
310 additions
and
301 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,207 +1 @@ | ||
from torch import nn | ||
from torch import Tensor | ||
from .utils import load_state_dict_from_url | ||
from typing import Callable, Any, Optional, List | ||
|
||
|
||
__all__ = ['MobileNetV2', 'mobilenet_v2'] | ||
|
||
|
||
model_urls = { | ||
'mobilenet_v2': 'https://download.pytorch.org/models/mobilenet_v2-b0353104.pth', | ||
} | ||
|
||
|
||
def _make_divisible(v: float, divisor: int, min_value: Optional[int] = None) -> int: | ||
""" | ||
This function is taken from the original tf repo. | ||
It ensures that all layers have a channel number that is divisible by 8 | ||
It can be seen here: | ||
https://github.com/tensorflow/models/blob/master/research/slim/nets/mobilenet/mobilenet.py | ||
:param v: | ||
:param divisor: | ||
:param min_value: | ||
:return: | ||
""" | ||
if min_value is None: | ||
min_value = divisor | ||
new_v = max(min_value, int(v + divisor / 2) // divisor * divisor) | ||
# Make sure that round down does not go down by more than 10%. | ||
if new_v < 0.9 * v: | ||
new_v += divisor | ||
return new_v | ||
|
||
|
||
class ConvBNReLU(nn.Sequential): | ||
def __init__( | ||
self, | ||
in_planes: int, | ||
out_planes: int, | ||
kernel_size: int = 3, | ||
stride: int = 1, | ||
groups: int = 1, | ||
norm_layer: Optional[Callable[..., nn.Module]] = None | ||
) -> None: | ||
padding = (kernel_size - 1) // 2 | ||
if norm_layer is None: | ||
norm_layer = nn.BatchNorm2d | ||
super(ConvBNReLU, self).__init__( | ||
nn.Conv2d(in_planes, out_planes, kernel_size, stride, padding, groups=groups, bias=False), | ||
norm_layer(out_planes), | ||
nn.ReLU6(inplace=True) | ||
) | ||
|
||
|
||
class InvertedResidual(nn.Module): | ||
def __init__( | ||
self, | ||
inp: int, | ||
oup: int, | ||
stride: int, | ||
expand_ratio: int, | ||
norm_layer: Optional[Callable[..., nn.Module]] = None | ||
) -> None: | ||
super(InvertedResidual, self).__init__() | ||
self.stride = stride | ||
assert stride in [1, 2] | ||
|
||
if norm_layer is None: | ||
norm_layer = nn.BatchNorm2d | ||
|
||
hidden_dim = int(round(inp * expand_ratio)) | ||
self.use_res_connect = self.stride == 1 and inp == oup | ||
|
||
layers: List[nn.Module] = [] | ||
if expand_ratio != 1: | ||
# pw | ||
layers.append(ConvBNReLU(inp, hidden_dim, kernel_size=1, norm_layer=norm_layer)) | ||
layers.extend([ | ||
# dw | ||
ConvBNReLU(hidden_dim, hidden_dim, stride=stride, groups=hidden_dim, norm_layer=norm_layer), | ||
# pw-linear | ||
nn.Conv2d(hidden_dim, oup, 1, 1, 0, bias=False), | ||
norm_layer(oup), | ||
]) | ||
self.conv = nn.Sequential(*layers) | ||
|
||
def forward(self, x: Tensor) -> Tensor: | ||
if self.use_res_connect: | ||
return x + self.conv(x) | ||
else: | ||
return self.conv(x) | ||
|
||
|
||
class MobileNetV2(nn.Module): | ||
def __init__( | ||
self, | ||
num_classes: int = 1000, | ||
width_mult: float = 1.0, | ||
inverted_residual_setting: Optional[List[List[int]]] = None, | ||
round_nearest: int = 8, | ||
block: Optional[Callable[..., nn.Module]] = None, | ||
norm_layer: Optional[Callable[..., nn.Module]] = None | ||
) -> None: | ||
""" | ||
MobileNet V2 main class | ||
Args: | ||
num_classes (int): Number of classes | ||
width_mult (float): Width multiplier - adjusts number of channels in each layer by this amount | ||
inverted_residual_setting: Network structure | ||
round_nearest (int): Round the number of channels in each layer to be a multiple of this number | ||
Set to 1 to turn off rounding | ||
block: Module specifying inverted residual building block for mobilenet | ||
norm_layer: Module specifying the normalization layer to use | ||
""" | ||
super(MobileNetV2, self).__init__() | ||
|
||
if block is None: | ||
block = InvertedResidual | ||
|
||
if norm_layer is None: | ||
norm_layer = nn.BatchNorm2d | ||
|
||
input_channel = 32 | ||
last_channel = 1280 | ||
|
||
if inverted_residual_setting is None: | ||
inverted_residual_setting = [ | ||
# t, c, n, s | ||
[1, 16, 1, 1], | ||
[6, 24, 2, 2], | ||
[6, 32, 3, 2], | ||
[6, 64, 4, 2], | ||
[6, 96, 3, 1], | ||
[6, 160, 3, 2], | ||
[6, 320, 1, 1], | ||
] | ||
|
||
# only check the first element, assuming user knows t,c,n,s are required | ||
if len(inverted_residual_setting) == 0 or len(inverted_residual_setting[0]) != 4: | ||
raise ValueError("inverted_residual_setting should be non-empty " | ||
"or a 4-element list, got {}".format(inverted_residual_setting)) | ||
|
||
# building first layer | ||
input_channel = _make_divisible(input_channel * width_mult, round_nearest) | ||
self.last_channel = _make_divisible(last_channel * max(1.0, width_mult), round_nearest) | ||
features: List[nn.Module] = [ConvBNReLU(3, input_channel, stride=2, norm_layer=norm_layer)] | ||
# building inverted residual blocks | ||
for t, c, n, s in inverted_residual_setting: | ||
output_channel = _make_divisible(c * width_mult, round_nearest) | ||
for i in range(n): | ||
stride = s if i == 0 else 1 | ||
features.append(block(input_channel, output_channel, stride, expand_ratio=t, norm_layer=norm_layer)) | ||
input_channel = output_channel | ||
# building last several layers | ||
features.append(ConvBNReLU(input_channel, self.last_channel, kernel_size=1, norm_layer=norm_layer)) | ||
# make it nn.Sequential | ||
self.features = nn.Sequential(*features) | ||
|
||
# building classifier | ||
self.classifier = nn.Sequential( | ||
nn.Dropout(0.2), | ||
nn.Linear(self.last_channel, num_classes), | ||
) | ||
|
||
# weight initialization | ||
for m in self.modules(): | ||
if isinstance(m, nn.Conv2d): | ||
nn.init.kaiming_normal_(m.weight, mode='fan_out') | ||
if m.bias is not None: | ||
nn.init.zeros_(m.bias) | ||
elif isinstance(m, (nn.BatchNorm2d, nn.GroupNorm)): | ||
nn.init.ones_(m.weight) | ||
nn.init.zeros_(m.bias) | ||
elif isinstance(m, nn.Linear): | ||
nn.init.normal_(m.weight, 0, 0.01) | ||
nn.init.zeros_(m.bias) | ||
|
||
def _forward_impl(self, x: Tensor) -> Tensor: | ||
# This exists since TorchScript doesn't support inheritance, so the superclass method | ||
# (this one) needs to have a name other than `forward` that can be accessed in a subclass | ||
x = self.features(x) | ||
# Cannot use "squeeze" as batch-size can be 1 => must use reshape with x.shape[0] | ||
x = nn.functional.adaptive_avg_pool2d(x, (1, 1)).reshape(x.shape[0], -1) | ||
x = self.classifier(x) | ||
return x | ||
|
||
def forward(self, x: Tensor) -> Tensor: | ||
return self._forward_impl(x) | ||
|
||
|
||
def mobilenet_v2(pretrained: bool = False, progress: bool = True, **kwargs: Any) -> MobileNetV2: | ||
""" | ||
Constructs a MobileNetV2 architecture from | ||
`"MobileNetV2: Inverted Residuals and Linear Bottlenecks" <https://arxiv.org/abs/1801.04381>`_. | ||
Args: | ||
pretrained (bool): If True, returns a model pre-trained on ImageNet | ||
progress (bool): If True, displays a progress bar of the download to stderr | ||
""" | ||
model = MobileNetV2(**kwargs) | ||
if pretrained: | ||
state_dict = load_state_dict_from_url(model_urls['mobilenet_v2'], | ||
progress=progress) | ||
model.load_state_dict(state_dict) | ||
return model | ||
from .mobilenetv2 import MobileNetV2, mobilenet_v2 |
Oops, something went wrong.