我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用torch.nn.Module()。
def _augment_module_post(net: nn.Module, callback_dict: dict) -> (dict, list): backward_hook_remove_func_list = [] vis_param_dict = dict() vis_param_dict['layer'] = None vis_param_dict['index'] = None vis_param_dict['method'] = GradType.NAIVE for x, y in net.named_modules(): if not isinstance(y, nn.Sequential) and y is not net: # I should add hook to all layers, in case they will be needed. backward_hook_remove_func_list.append( y.register_backward_hook( partial(_backward_hook, module_name=x, callback_dict=callback_dict, vis_param_dict=vis_param_dict))) def remove_handles(): for x in backward_hook_remove_func_list: x.remove() return vis_param_dict, remove_handles
def generate_model(): class MyModel(nn.Module): def __init__(self, pretrained_model): super(MyModel, self).__init__() self.pretrained_model = pretrained_model self.layer1 = pretrained_model.layer1 self.layer2 = pretrained_model.layer2 self.layer3 = pretrained_model.layer3 self.layer4 = pretrained_model.layer4 pretrained_model.avgpool = nn.AvgPool2d(8) classifier = [ nn.Linear(pretrained_model.fc.in_features, 17), ] self.classifier = nn.Sequential(*classifier) pretrained_model.fc = self.classifier def forward(self, x): return F.sigmoid(self.pretrained_model(x)) return MyModel(torchvision.models.resnet50(pretrained=True))
def generate_model(): class MyModel(nn.Module): def __init__(self, pretrained_model): super(MyModel, self).__init__() self.pretrained_model = pretrained_model self.layer1 = pretrained_model.layer1 self.layer2 = pretrained_model.layer2 self.layer3 = pretrained_model.layer3 self.layer4 = pretrained_model.layer4 pretrained_model.avgpool = nn.AvgPool2d(8) classifier = [ nn.Linear(pretrained_model.fc.in_features, 17), ] self.classifier = nn.Sequential(*classifier) pretrained_model.fc = self.classifier def forward(self, x): return F.sigmoid(self.pretrained_model(x)) return MyModel(torchvision.models.resnet34(pretrained=True))
def generate_model(): class MyModel(nn.Module): def __init__(self, pretrained_model): super(MyModel, self).__init__() self.pretrained_model = pretrained_model self.layer1 = pretrained_model.layer1 self.layer2 = pretrained_model.layer2 self.layer3 = pretrained_model.layer3 self.layer4 = pretrained_model.layer4 pretrained_model.avgpool = nn.AvgPool2d(8) classifier = [ nn.Linear(pretrained_model.fc.in_features, 17), ] self.classifier = nn.Sequential(*classifier) pretrained_model.fc = self.classifier def forward(self, x): return F.sigmoid(self.pretrained_model(x)) return MyModel(torchvision.models.resnet101(pretrained=True))
def generate_model(): class MyModel(nn.Module): def __init__(self, pretrained_model): super(MyModel, self).__init__() self.pretrained_model = pretrained_model self.layer1 = pretrained_model.layer1 self.layer2 = pretrained_model.layer2 self.layer3 = pretrained_model.layer3 self.layer4 = pretrained_model.layer4 pretrained_model.avgpool = nn.AvgPool2d(8) classifier = [ nn.Linear(pretrained_model.fc.in_features, 17), ] self.classifier = nn.Sequential(*classifier) pretrained_model.fc = self.classifier def forward(self, x): return self.pretrained_model(x) return MyModel(torchvision.models.resnet18(pretrained=True))
def call_nn_op(op, epsilon): """ a helper function that adds appropriate parameters when calling an nn module representing an operation like Softmax :param op: the nn.Module operation to instantiate :param epsilon: a scaling parameter for certain custom modules :return: instantiation of the op module with appropriate parameters """ if op in [ClippedSoftmax]: try: return op(epsilon, dim=1) except TypeError: # Support older pytorch 0.2 release. return op(epsilon) elif op in [ClippedSigmoid]: return op(epsilon) elif op in [nn.Softmax, nn.LogSoftmax]: return op(dim=1) else: return op()
def setUp(self): import torch.nn as nn from inferno.utils.python_utils import from_iterable class DummyNamedModule(nn.Module): def __init__(self, name, history, num_inputs=1): super(DummyNamedModule, self).__init__() self.name = name self.history = history self.num_inputs = num_inputs def forward(self, *inputs): assert len(inputs) == self.num_inputs self.history.append(self.name) if self.num_inputs > 1: output = reduce(lambda x, y: x + y, inputs) else: output = from_iterable(inputs) return output self.DummyNamedModule = DummyNamedModule # @unittest.skip
def get_module_for_nodes(self, names): """ Gets the `torch.nn.Module` object for nodes corresponding to `names`. Parameters ---------- names : str or list of str Names of the nodes to fetch the modules of. Returns ------- list or torch.nn.Module Module or a list of modules corresponding to `names`. """ names = pyu.to_iterable(names) modules = [] for name in names: assert self.is_node_in_graph(name), "Node '{}' is not in graph.".format(name) module = getattr(self, name, None) assert module is not None, "Node '{}' is in the graph but could not find a module " \ "corresponding to it.".format(name) modules.append(module) return pyu.from_iterable(modules)
def header_code(self): return """import numpy as np import torch import torch.nn as nn import torch.nn.functional as F __weights_dict = dict() def load_weights(weight_file): if weight_file == None: return try: weights_dict = np.load(weight_file).item() except: weights_dict = np.load(weight_file, encoding='bytes').item() return weights_dict class KitModel(nn.Module): """
def new_proj_module(self): emb_dim = self.emb_dim mem_dim = self.mem_dim class NewProjModule(nn.Module): def __init__(self, emb_dim, mem_dim): super(NewProjModule, self).__init__() self.emb_dim = emb_dim self.mem_dim = mem_dim self.linear1 = nn.Linear(self.emb_dim, self.mem_dim) self.linear2 = nn.Linear(self.emb_dim, self.mem_dim) def forward(self, input): i = nn.Sigmoid()(self.linear1(input)) u = nn.Tanh()(self.linear2(input)) out = i.mul(u) # CMulTable().updateOutput([i, u]) return out module = NewProjModule(emb_dim, mem_dim) # if getattr(self, "proj_module_master", None): # share parameters # for (tar_param, src_param) in zip(module.parameters(), self.proj_module_master.parameters()): # tar_param.grad.data = src_param.grad.data.clone() return module
def new_att_module(self): class NewAttModule(nn.Module): def __init__(self): super(NewAttModule, self).__init__() def forward(self, linput, rinput): self.lPad = linput.view(-1, linput.size(0), linput.size(1)) self.lPad = linput # self.lPad = Padding(0, 0)(linput) TODO: figureout why padding? self.M_r = torch.mm(self.lPad, rinput.t()) self.alpha = F.softmax(self.M_r.transpose(0, 1)) self.Yl = torch.mm(self.alpha, self.lPad) return self.Yl att_module = NewAttModule() if getattr(self, "att_module_master", None): for (tar_param, src_param) in zip(att_module.parameters(), self.att_module_master.parameters()): tar_param.grad.data = src_param.grad.data.clone() return att_module
def register_parameter(self, name, param): """Adds a parameter to the module. The parameter can be accessed as an attribute using given name. """ if '_parameters' not in self.__dict__: raise AttributeError( "cannot assign parameter before Module.__init__() call") if param is None: self._parameters[name] = None elif not isinstance(param, Parameter): raise TypeError("cannot assign '{}' object to parameter '{}' " "(torch.nn.Parameter or None required)" .format(torch.typename(param), name)) elif param.grad_fn: raise ValueError( "Cannot assign non-leaf Variable to parameter '{0}'. Model " "parameters must be created explicitly. To express '{0}' " "as a function of another variable, compute the value in " "the forward() method.".format(name)) else: self._parameters[name] = param
def test_parameters(self): def num_params(module): return len(list(module.parameters())) class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.l1 = l self.l2 = l self.param = Parameter(torch.Tensor(3, 5)) l = nn.Linear(10, 20) n = Net() s = nn.Sequential(n, n, n, n) self.assertEqual(num_params(l), 2) self.assertEqual(num_params(n), 3) self.assertEqual(num_params(s), 3)
def test_named_modules(self): class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.l1 = l self.l2 = l self.param = Variable(torch.Tensor(3, 5)) self.block = block l = nn.Linear(10, 20) l1 = nn.Linear(10, 20) l2 = nn.Linear(10, 20) block = nn.Sequential() block.add_module('linear1', l1) block.add_module('linear2', l2) n = Net() s = nn.Sequential(n, n, n, n) self.assertEqual(list(s.named_modules()), [('', s), ('0', n), ('0.l1', l), ('0.block', block), ('0.block.linear1', l1), ('0.block.linear2', l2)])
def test_type(self): l = nn.Linear(10, 20) net = nn.Module() net.l = l net.l2 = l net.add_module('empty', None) net.float() self.assertIsInstance(l.weight.data, torch.FloatTensor) self.assertIsInstance(l.bias.data, torch.FloatTensor) net.double() self.assertIsInstance(l.weight.data, torch.DoubleTensor) self.assertIsInstance(l.bias.data, torch.DoubleTensor) net.type(torch.FloatTensor) self.assertIsInstance(l.weight.data, torch.FloatTensor) self.assertIsInstance(l.bias.data, torch.FloatTensor) net.type(torch.DoubleTensor) self.assertIsInstance(l.weight.data, torch.DoubleTensor) self.assertIsInstance(l.bias.data, torch.DoubleTensor) if TEST_CUDA: net.type(torch.cuda.FloatTensor) self.assertIsInstance(l.weight.data, torch.cuda.FloatTensor) self.assertIsInstance(l.bias.data, torch.cuda.FloatTensor)
def test_data_parallel_nested_output(self): def fn(input): return [input, (input.sin(), input.cos(), [input.add(1)]), input] class Net(nn.Module): def forward(self, input): return fn(input) i = Variable(torch.randn(2, 2).float().cuda(1)) gpus = range(torch.cuda.device_count()) output = dp.data_parallel(Net(), i, gpus) self.assertEqual(output, fn(i)) self.assertIsInstance(output[0], Variable) self.assertIsInstance(output[1], tuple) self.assertIsInstance(output[1][0], Variable) self.assertIsInstance(output[1][1], Variable) self.assertIsInstance(output[1][2], list) self.assertIsInstance(output[1][2][0], Variable) self.assertIsInstance(output[2], Variable)
def test_container_copy(self): class Model(nn.Module): def __init__(self): super(Model, self).__init__() self.linear = nn.Linear(4, 5) def forward(self, input): return self.linear(input) input = Variable(torch.randn(2, 4)) model = Model() model_cp = deepcopy(model) self.assertEqual(model(input).data, model_cp(input).data) model_cp.linear.weight.data[:] = 2 self.assertNotEqual(model(input).data, model_cp(input).data)
def test_mini_wlm(self): """Exercise null-edge pruning in the tracer.""" @torch.jit.compile(verify=True) class MyModel(nn.Module): def __init__(self): super(MyModel, self).__init__() self.encoder = nn.Embedding(2, 2) def forward(self, input, hidden): emb = self.encoder(input) hidden = hidden.clone() # simulate some RNN operation return emb, hidden model = MyModel() x = Variable(torch.LongTensor([[0, 1], [1, 0]])) y = Variable(torch.FloatTensor([0])) z, _ = model(x, y) z.sum().backward() z, _ = model(x, y) z.sum().backward()
def test_data_parallel_module_kwargs_only(self): class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.l = l def forward(self, input): return self.l(input) l = nn.Linear(10, 5).float().cuda() i = Variable(torch.randn(20, 10).float().cuda()) expected_out = l(i).data n = nn.DataParallel(Net()) out = n(input=i) self.assertEqual(out.get_device(), 0) self.assertEqual(out.data, expected_out)
def register_parameter(self, name, param, bounds, prior=None): """ Adds a parameter to the module. The parameter can be accessed as an attribute using given name. name (str): name of parameter param (torch.nn.Parameter): parameter bounds (2-tuple of float or Tensor): lower and upper bounds for parameter prior (RandomVariable): prior for parameter (default: None) """ if '_parameters' not in self.__dict__: raise AttributeError( "cannot assign parameter before Module.__init__() call") super(Module, self).register_parameter(name, param) kwargs = {} kwargs[name] = bounds self.set_bounds(**kwargs)
def __init__(self, kernel_size=11, log_t=False): """ Module which Performs a single attention step along the second axis of a given encoded input. The module uses both 'content' and 'location' based attention. The 'content' based attention is an inner product of the decoder hidden state with each time-step of the encoder state. The 'location' based attention performs a 1D convollution on the previous attention vector and adds this into the next attention vector prior to normalization. *NB* Should compute attention differently if using cuda or cpu based on performance. See https://gist.github.com/awni/9989dd31642d42405903dec8ab91d1f0 """ super(Attention, self).__init__() assert kernel_size % 2 == 1, \ "Kernel size should be odd for 'same' conv." padding = (kernel_size - 1) // 2 self.conv = nn.Conv1d(1, 1, kernel_size, padding=padding) self.log_t = log_t
def train(self, mode=True): # Override train so that the training mode is set as we want nn.Module.train(self, mode) if mode: # Set fixed blocks to be in eval mode self.RCNN_base.eval() self.RCNN_base[5].train() self.RCNN_base[6].train() def set_bn_eval(m): classname = m.__class__.__name__ if classname.find('BatchNorm') != -1: m.eval() self.RCNN_base.apply(set_bn_eval) self.RCNN_top.apply(set_bn_eval)
def add_module(self, name, module): """Adds a child module to the current module. The module can be accessed as an attribute using the given name. Args: name (string): name of the child module. The child module can be accessed from this module using the given name parameter (Module): child module to be added to the module. """ if not isinstance(module, Module) and module is not None: raise TypeError("{} is not a Module subclass".format( torch.typename(module))) if hasattr(self, name) and name not in self._modules: raise KeyError("attribute '{}' already exists".format(name)) self._modules[name] = module
def named_children(self): """Returns an iterator over immediate children modules, yielding both the name of the module as well as the module itself. Yields: (string, Module): Tuple containing a name and child module Example: >>> for name, module in model.named_children(): >>> if name in ['conv4', 'conv5']: >>> print(module) """ memo = set() for name, module in self._modules.items(): if module is not None and module not in memo: memo.add(module) yield name, module
def modules(self): """Returns an iterator over all modules in the network. Yields: Module: a module in the network Note: Duplicate modules are returned only once. In the following example, ``l`` will be returned only once. >>> l = nn.Linear(2, 2) >>> net = nn.Sequential(l, l) >>> for idx, m in enumerate(net.modules()): >>> print(idx, '->', m) 0 -> Sequential ( (0): Linear (2 -> 2) (1): Linear (2 -> 2) ) 1 -> Linear (2 -> 2) """ for name, module in self.named_modules(): yield module
def test_call_supports_python_dict_output(self): class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.l1 = nn.Linear(10, 20) self.register_backward_hook(self.hook) self.check_backward_hook_flag = False def hook(self, module, grad_out, grad_in): self.check_backward_hook_flag = True def forward(self, inputs): return {"output": self.l1(inputs).sum()} net = Net() model_output = net(Variable(torch.randn([5, 10]))) model_output["output"].backward() self.assertTrue(net.check_backward_hook_flag)
def test_data_parallel_module_kwargs_only_empty_list(self): class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.l = l def forward(self, input): return self.l(input['data']) l = nn.Linear(10, 5).float().cuda() i = Variable(torch.randn(20, 10).float().cuda()) expected_out = l(i).data n = nn.DataParallel(Net()) out = n(input={'data': i, 'unused': []}) self.assertEqual(out.get_device(), 0) self.assertEqual(out.data, expected_out)
def test_data_parallel_module_kwargs_only_empty_dict(self): class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.l = l def forward(self, input): return self.l(input['data']) l = nn.Linear(10, 5).float().cuda() i = Variable(torch.randn(20, 10).float().cuda()) expected_out = l(i).data n = nn.DataParallel(Net()) out = n(input={'data': i, 'unused': {}}) self.assertEqual(out.get_device(), 0) self.assertEqual(out.data, expected_out)
def test_data_parallel_module_kwargs_only_empty_tuple(self): class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.l = l def forward(self, input): return self.l(input['data']) l = nn.Linear(10, 5).float().cuda() i = Variable(torch.randn(20, 10).float().cuda()) expected_out = l(i).data n = nn.DataParallel(Net()) out = n(input={'data': i, 'unused': ()}) self.assertEqual(out.get_device(), 0) self.assertEqual(out.data, expected_out)
def num_flat_features(self, x): size = x.size()[1:] # all dimensions except the batch dimension num_features = 1 for s in size: num_features *= s return num_features # class NNetM(nn.Module): # # def __init__(self, n_in, n_out): # super(NNetM, self).__init__() # # self.fc1 = nn.Linear(n_in, 120) # self.fc2 = nn.Linear(120, 84) # self.fc3 = nn.Linear(84, n_out[0]*n_out[1]) # # def forward(self, x): # # x = F.relu(self.fc1(x)) # x = F.relu(self.fc2(x)) # x = self.fc3(x) # return x
def block(self, name, in_channels, out_channels, pool_stride=2): """ Stack n bottleneck modules where n is inferred from the depth of the network. Args: name: string name of the current block. in_channels: number of input channels out_channels: number of output channels pool_stride: factor to reduce the spatial dimensionality in the first bottleneck of the block. Returns: a Module consisting of n sequential bottlenecks. """ block = nn.Sequential() for bottleneck in range(self.block_depth): name_ = '%s_bottleneck_%d' % (name, bottleneck) if bottleneck == 0: block.add_module(name_, ScaleResNeXtBottleneck(in_channels, out_channels, pool_stride, self.cardinality, self.widen_factor)) else: block.add_module(name_, ScaleResNeXtBottleneck(out_channels, out_channels, 1, self.cardinality, self.widen_factor)) return block
def block(self, name, in_channels, out_channels, pool_stride=2): """ Stack n bottleneck modules where n is inferred from the depth of the network. Args: name: string name of the current block. in_channels: number of input channels out_channels: number of output channels pool_stride: factor to reduce the spatial dimensionality in the first bottleneck of the block. Returns: a Module consisting of n sequential bottlenecks. """ block = nn.Sequential() for bottleneck in range(self.block_depth): name_ = '%s_bottleneck_%d' % (name, bottleneck) if bottleneck == 0: block.add_module(name_, ResNeXtBottleneck(in_channels, out_channels, pool_stride, self.cardinality, self.widen_factor)) else: block.add_module(name_, ResNeXtBottleneck(out_channels, out_channels, 1, self.cardinality, self.widen_factor)) return block
