python类links()的实例源码

def _set_module(self, namespace, module):
        assert isinstance(module, Module)

        for index, layer in enumerate(module.layers):
            if isinstance(layer, chainer.Link):
                super(Module, self).__setattr__("_module_{}_sequential_{}".format(namespace, index), layer)

            if isinstance(layer, Residual):
                for _index, _layer in enumerate(layer.layers):
                    if isinstance(_layer, chainer.Link):
                        super(Module, self).__setattr__("_module_{}_sequential_{}_{}".format(namespace, index, _index), _layer)

        for index, (link_name, link) in enumerate(module.links):
            assert isinstance(link, chainer.Link)
            super(Module, self).__setattr__("_module_{}_link_{}".format(namespace, link_name), link)

        for index, (module_name, module) in enumerate(module.modules):
            assert isinstance(module, Module)
            self._set_module("{}_{}".format(namespace, module_name), module)

        module._locked = True

def test_convolution(self):
        self.init_func()
        self.assertEqual(len(self.func.layers), 1)
        f = self.func.l1
        self.assertIsInstance(f, links.Convolution2D)
        for i in range(3):  # 3 == group
            in_slice = slice(i * 4, (i + 1) * 4)  # 4 == channels
            out_slice = slice(i * 2, (i + 1) * 2)  # 2 == num / group
            w = f.W.data[out_slice, in_slice]
            numpy.testing.assert_array_equal(
                w.flatten(), range(i * 32, (i + 1) * 32))

        numpy.testing.assert_array_equal(
            f.b.data, range(6))

        self.call(['x'], ['y'])
        self.mock.assert_called_once_with(self.inputs[0])

def __init__(self, n_layers, n_units, width=3, dropout=0.2):
        super(ConvGLUDecoder, self).__init__()
        links = [('l{}'.format(i + 1),
                  ConvGLU(n_units, width=width,
                          dropout=dropout, nopad=True))
                 for i in range(n_layers)]
        for link in links:
            self.add_link(*link)
        self.conv_names = [name for name, _ in links]
        self.width = width

        init_preatt = VarInNormal(1.)
        links = [('preatt{}'.format(i + 1),
                  L.Linear(n_units, n_units, initialW=init_preatt))
                 for i in range(n_layers)]
        for link in links:
            self.add_link(*link)
        self.preatt_names = [name for name, _ in links]

def __init__(self, layer, in_size, ch, out_size, stride=2):
        super(Block, self).__init__()
        links = [('a', BottleNeckA(in_size, ch, out_size, stride))]
        for i in range(layer - 1):
            links += [('b_{}'.format(i + 1), BottleNeckB(out_size, ch))]

        for link in links:
            self.add_link(*link)
        self.forward = links

def create_acer_agent(env):
    obs_dim = env.observation_space.shape[0]
    n_actions = env.action_space.n

    model = acer.ACERSeparateModel(
        pi=links.Sequence(
            L.Linear( obs_dim, 1024, initialW=LeCunNormal(1e-3)),
            F.relu,
            L.Linear( 1024, 512, initialW=LeCunNormal(1e-3)),
            F.relu,
            L.Linear( 512, n_actions, initialW=LeCunNormal(1e-3)),
            SoftmaxDistribution),
        q=links.Sequence(
            L.Linear( obs_dim, 1024, initialW=LeCunNormal(1e-3)),
            F.relu,
            L.Linear( 1024, 512, initialW=LeCunNormal(1e-3)),
            F.relu,
            L.Linear( 512, n_actions, initialW=LeCunNormal(1e-3)),
            DiscreteActionValue),
        )

    opt = rmsprop_async.RMSpropAsync( lr=7e-4, eps=1e-2, alpha=0.99)
    opt.setup( model )
    opt.add_hook( chainer.optimizer.GradientClipping(40) )

    replay_buffer = EpisodicReplayBuffer( 128 )
    agent = acer.ACER( model, opt, 
        gamma=0.95, # reward discount factor
        t_max=32, # update the model after this many local steps
        replay_buffer=replay_buffer,
        n_times_replay=4, # number of times experience replay is repeated for each update
        replay_start_size=64, # don't start replay unless we have this many experiences in the buffer
        disable_online_update=True, # rely only on experience buffer
        use_trust_region=True,  # enable trust region policy optimiztion
        trust_region_delta=0.1,  # a parameter for TRPO
        truncation_threshold=5.0, # truncate large importance weights
        beta=1e-2, # entropy regularization parameter
        phi= lambda obs: obs.astype(np.float32, copy=False) )

    return agent

def remove_link(self, name):
        """Remove a link that has the given name from this model

        Optimizer sees ``~Chain.namedparams()`` to know which parameters should
        be updated. And inside of ``namedparams()``, ``self._children`` is
        called to get names of all links included in the Chain.
        """
        self._children.remove(name)

def convert_debug(self, content_img, init_img, output_directory, max_iteration=1000, debug_span=100,
                      random_init=False):
        init_array = self.xp.array(neural_art.utility.img2array(init_img))
        if random_init:
            init_array = self.xp.array(self.xp.random.uniform(-20, 20, init_array.shape), dtype=init_array.dtype)
        content_array = self.xp.array(neural_art.utility.img2array(content_img))
        content_layers = self.model.forward_layers(chainer.Variable(content_array),
                                                   average_pooling=self.average_pooling)

        parameter_now = chainer.links.Parameter(init_array)
        self.optimizer.setup(parameter_now)
        for i in range(max_iteration + 1):
            neural_art.utility.print_ltsv({"iteration": i})
            if i % debug_span == 0 and i > 0:
                print("dump to {}".format(os.path.join(output_directory, "{}.png".format(i))))
                neural_art.utility.array2img(chainer.cuda.to_cpu(parameter_now.W.data)).save(
                    os.path.join(output_directory, "{}.png".format(i)))
            parameter_now.zerograds()
            x = parameter_now.W
            layers = self.model.forward_layers(x, average_pooling=self.average_pooling)

            loss_texture = self._texture_loss(layers)
            loss_content = self._contents_loss(layers, content_layers)
            loss = self.texture_weight * loss_texture + self.content_weight * loss_content
            loss.backward()
            parameter_now.W.grad = x.grad
            self.optimizer.update()
        return neural_art.utility.array2img(chainer.cuda.to_cpu(parameter_now.W.data))

def __call__(self, x, train=True):
        h = x
        links = self.children()
        for link in links:
            h = link(h, train=train)
        return h

def __call__(self, x, train=True):
        links = self.children()
        h = F.leaky_relu(next(links)(x))
        for link in links:
            h = link(h, train)
        return h

def Convolution2D(in_channel, out_channel, ksize, stride=1, pad=0, initialW=None, weightnorm=False):
    if weightnorm:
        return WeightnormConvolution2D(in_channel, out_channel, ksize, stride=1, pad=pad, initialV=initialW)
    return links.Convolution2D(in_channel, out_channel, ksize, stride=1, pad=pad, initialW=initialW)

def __init__(self, *layers):
        super(Module, self).__init__()
        self.layers = []
        self.blocks = []
        self.links = []
        self.modules = []
        self._locked = False
        if len(layers) > 0:
            self.add(*layers)

def __setattr__(self, name, value):
        if isinstance(value, Module):
            self.modules.append((name, value))
            self._set_module(name, value)
            return super(chainer.Link, self).__setattr__(name, value)   # prevent module from being added to self._children

        if isinstance(value, chainer.Link):
            assert self._locked is False, "Since this module is owned by another module, it is not possible to add Link."
            with self.init_scope():
                if name.startswith("_sequential_"):
                    return super(Module, self).__setattr__(name, value)
                self.links.append((name, value))
                return super(Module, self).__setattr__(name, value)

        super(Module, self).__setattr__(name, value)

def __init__(self, in_channels, out_channels, kernel_size=2, pooling="f", zoneout=False, wgain=1., weightnorm=False):
        super(QRNNDecoder, self).__init__(in_channels, out_channels, kernel_size, pooling, zoneout, wgain, weightnorm)
        self.num_split = len(pooling) + 1
        wstd = math.sqrt(wgain / in_channels / kernel_size)

        with self.init_scope():
            setattr(self, "V", links.Linear(out_channels, self.num_split * out_channels, initialW=initializers.Normal(wstd)))

    # ht_enc is the last encoder state

def __init__(self, in_channels, out_channels, kernel_size=2, zoneout=False, wgain=1., weightnorm=False):
        super(QRNNGlobalAttentiveDecoder, self).__init__(in_channels, out_channels, kernel_size, "fo", zoneout, wgain, weightnorm)
        wstd = math.sqrt(wgain / in_channels / kernel_size)
        with self.init_scope():
            setattr(self, "o", links.Linear(2 * out_channels, out_channels, initialW=initializers.Normal(wstd)))

    # X is the input of the decoder
    # ht_enc is the last encoder state
    # H_enc is the encoder's las layer's hidden sates

def links(self, skipself=False):
        """Returns a generator of all links under the hierarchy.

        Args:
            skipself (bool): If ``True``, then the generator skips this link
                and starts with the first child link.

        Returns:
            A generator object that generates all links.

        """
        if not skipself:
            yield self

def children(self):
        """Returns a generator of all child links.

        Returns:
            A generator object that generates all child links.

        """
        if 0:
            yield

def __init__(self, **links):
        super(Chain, self).__init__()
        self._children = []

        for name, link in six.iteritems(links):
            self.add_link(name, link)

def copy(self):
        ret = super(Chain, self).copy()
        ret._children = list(ret._children)
        d = ret.__dict__
        for name in ret._children:
            # copy child links recursively
            copied = d[name].copy()
            copied.name = name
            d[name] = copied
        return ret

def links(self, skipself=False):
        if not skipself:
            yield self
        d = self.__dict__
        for name in self._children:
            for link in d[name].links():
                yield link

def links(self, skipself=False):
        if not skipself:
            yield self
        for child in self._children:
            for link in child.links():
                yield link

def test_linear(self):
        self.init_func()
        f = self.func.l1
        self.assertIsInstance(f, links.Linear)
        numpy.testing.assert_array_equal(
            f.W.data, numpy.array([[0, 1, 2], [3, 4, 5]], dtype=numpy.float32))
        numpy.testing.assert_array_equal(
            f.b.data, numpy.array([0, 1], dtype=numpy.float32))

        self.call(['x'], ['y'])
        self.mock.assert_called_once_with(self.inputs[0])

def test_linear(self):
        self.init_func()
        f = self.func.l1
        self.assertIsInstance(f, links.Linear)
        numpy.testing.assert_array_equal(
            f.W.data, numpy.array([[0, 1, 2], [3, 4, 5]], dtype=numpy.float32))
        self.assertIsNone(f.b)

        self.call(['x'], ['y'])
        self.mock.assert_called_once_with(self.inputs[0])

def test_py2_available(self):
        self.assertTrue(links.caffe.caffe_function.available)

def __init__(self, out_dims=64, normalize_output=False):
        super(ModifiedGoogLeNet, self).__init__()
        # remove links and functions
        for name in [n for n in self._children if n.startswith('loss')]:
            self._children.remove(name)
            delattr(self, name)
        self.functions.pop('loss3_fc')
        self.functions.pop('prob')

        self.add_link('bn_fc', L.BatchNormalization(1024))
        self.add_link('fc', L.Linear(1024, out_dims))

        image_mean = np.array([123, 117, 104], dtype=np.float32)  # RGB
        self._image_mean = image_mean[None, :, None, None]
        self.normalize_output = normalize_output

def __init__(self, n_layers, n_units, width=3, dropout=0.2):
        super(ConvGLUEncoder, self).__init__()
        links = [('l{}'.format(i + 1),
                  ConvGLU(n_units, width=width, dropout=dropout))
                 for i in range(n_layers)]
        for link in links:
            self.add_link(*link)
        self.conv_names = [name for name, _ in links]

def __init__(self, layer, in_size, ch, out_size, stride=2, act=F.elu):
        super(Block, self).__init__()
        links = [('a', BottleNeckA(in_size, ch, out_size, stride, act))]
        for i in range(layer-1):
            links += [('b{}'.format(i+1), BottleNeckB(out_size, ch, act))]

        for link in links:
            self.add_link(*link)
        self.forward = links

def __init__(self, n_point, n_stage):
        super(CPM, self).__init__()
        self.add_link('branch', Branch())
        self.add_link('stage1', Stage1(n_point))
        links = []
        for i in xrange(n_stage-1):
            links += [('stage{}'.format(i+2), StageN(n_point))]
        for l in links:
            self.add_link(*l)

        self.forward = links
        self.train = True

def __init__(self, n_point, n_stage):
        super(CPM, self).__init__()
        self.add_link('branch', Branch())
        self.add_link('stage1', Stage1(n_point))
        links = []
        for i in xrange(n_stage-1):
            links += [('stage{}'.format(i+2), StageN(n_point))]
        for l in links:
            self.add_link(*l)

        self.forward = links
        self.train = True

def charRNN(self, context):  # input a list of word ids, output a list of word embeddings
        # if chainer.config.train:
        #     print("train")
        # else:
        #     print("test")
        contexts2charIds = self.index2charIds[context]

        #sorting the context_char, make sure array length in descending order
        # ref: https://docs.chainer.org/en/stable/reference/generated/chainer.links.LSTM.html?highlight=Variable-length
        context_char_length = np.array([len(t) for t in contexts2charIds])
        argsort = context_char_length.argsort()[::-1] # descending order
        argsort_reverse = np.zeros(len(argsort), dtype=np.int32)  # this is used to restore the original order
        for i in range(len(argsort)):
            argsort_reverse[argsort[i]] = i
        contexts2charIds = contexts2charIds[context_char_length.argsort()[::-1]]

        #transpose a 2D list/numpy array
        rnn_inputs = [[] for i in range(len(contexts2charIds[0]))]
        for j in range(len(contexts2charIds)) :
            for i in range(len(contexts2charIds[j])):
                rnn_inputs[i].append(contexts2charIds[j][i])

        self.reset_state()
        for i in range(len(rnn_inputs)):
            y_ = self(np.array(rnn_inputs[i], np.int32))
        y = self.out(self.mid.h)
        y = y[argsort_reverse] # restore the original order
        return y

def charRNN(self, context):  # input a list of word ids, output a list of word embeddings
        # if chainer.config.train:
        #     print("train")
        # else:
        #     print("test")
        contexts2charIds = index2charIds[context]

        #sorting the context_char, make sure array length in descending order
        # ref: https://docs.chainer.org/en/stable/reference/generated/chainer.links.LSTM.html?highlight=Variable-length
        context_char_length = np.array([len(t) for t in contexts2charIds])
        argsort = context_char_length.argsort()[::-1] # descending order
        argsort_reverse = np.zeros(len(argsort), dtype=np.int32)  # this is used to restore the original order
        for i in range(len(argsort)):
            argsort_reverse[argsort[i]] = i
        contexts2charIds = contexts2charIds[context_char_length.argsort()[::-1]]

        #transpose a 2D list/numpy array
        rnn_inputs = [[] for i in range(len(contexts2charIds[0]))]
        for j in range(len(contexts2charIds)) :
            for i in range(len(contexts2charIds[j])):
                rnn_inputs[i].append(contexts2charIds[j][i])

        self.reset_state()
        for i in range(len(rnn_inputs)):
            y_ = self(np.array(rnn_inputs[i], np.int32))
        y = self.out(self.mid.h)
        y = y[argsort_reverse] # restore the original order
        return y