blocks.py

Extension packages to Bayes Blocks library, part 1

# -*- coding: iso-8859-1 -*-
#
# Copyright (C) 2001-2006 Markus Harva, Antti Honkela, Alexander
# Ilin, Tapani Raiko, Harri Valpola and Tomas 謘tman.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License (included in file License.txt in the
# program package) for more details.
#
# $Id: blocks.py,v 1.1.1.1 2006/11/23 09:42:07 mha Exp $
#

from bblocks.Label import *
from bblocks import Helpers
from bblocks.PyNet import PyNet, PyNodeFactory

import Numeric, MLab
import math
import random

import util

class Constants:
    def __init__(self, net):
        nf = PyNodeFactory(net)
        self.c0 = nf.GetConstant("c0", 0)
        self.c1 = nf.GetConstant("c1", 1)
        self.c_5 = nf.GetConstant("c_5", -5)
        self.c_7 = nf.GetConstant("c_7", -7)

class Priors:
    def __init__(self, net, dim, labelbase):
        nf = PyNodeFactory(net)
        c0, c_7 = map(net.GetNode, ["c0", "c_7"])

        m = nf.GetGaussian("m" + labelbase, c0, c_7)
        v = nf.GetGaussian("v" + labelbase, c0, c_7)
        nf.EvidenceNode(v, mean=0, var=1, decay=1)
        m.SetPersist(5)
        v.SetPersist(5)

        outputs = []

        for i in range(dim):
            outputs.append(nf.GetGaussian(Label(labelbase, i), m, v))
            outputs[i].SetPersist(5)
            nf.EvidenceNode(outputs[i], mean=0, var=1, decay=1)

        self.outputs = outputs

class LinearMap:
    def __init__(self, net, indim, outdim, labelbase, initvalues = None):
        self.net = net
        self.nf = PyNodeFactory(net)
        self.rows = outdim
        self.cols = indim
        self.labelbase = labelbase
        self.initvalues = initvalues

        self.__buildnet()

    def __buildnet(self):
        c0 = self.net.GetNode("c0")
        for i in range(self.rows):
            for j in range(self.cols):
                a = self.nf.GetGaussian(self.coeff_label(i, j), c0, c0)
                a.SetPersist(5)
                if self.initvalues:
                    self.nf.EvidenceNode(a, mean = self.initvalues[i,j])
        
    def reinit(self, decay = 5):
        for i in range(self.rows):
            for j in range(self.cols):
                a = self.getweight(i, j)
                if init:
                    self.nf.EvidenceNode(a, mean = self.initvalues[ir,ic],
                                          decay = decay)
                else:
                    self.nf.EvidenceNode(a, mean = random.gauss(0, 0.4),
                                          decay = decay)

    def getweight(self, i, j):
        return self.net.GetVariable(self.coeff_label(i, j))

    def coeff_label(self, i, j):
        return Label(self.labelbase + "_a", i, j)

    def addweight(self, i, j, addcost = 0):
        input = self.net.GetNode(self.inlabels[j])
        output = self.net.GetNode(self.outlabels[i])

        if ((self.getweight(i, j) is not None) or (input is None)
            or (output is None)):
            return 
        
        c0 = self.net.GetNode("c0")
        w = Helpers.AddWeight(
            self.net, input, output, self.coeff_label(i, j), c0, c0,
            addcost, self.labelbase)

        return w

    def addweights(self, addcost = 0):
        """Tries to add weigths to all possible positions."""
        print "Adding weights to linear map %s..." % self.labelbase
        addedlabels = []
        for i in range(self.rows):
            for j in range(self.cols):
                w = self.addweight(i, j, addcost)
                if w is not None:
                    label = w.GetLabel()
                    print "Added %s" % label
                    addedlabels.append(label)
        return addedlabels

    def prune(self, addcost = 0):
        print "Pruning linear map %s..." % self.labelbase
        prunedlabels = []
        for i in range(self.rows):
            for j in range(self.cols):
                label = self.coeff_label(i, j)
                w = self.getweight(i, j)
                if (w is not None) and self.net.TryPruning(w, 1, addcost):
                    prunedlabels.append(label)
        return prunedlabels

    def linmap(self, inputs):
        assert len(inputs) == self.cols

        lb = self.labelbase
        sum = []

        for i in range(self.rows):
            prod = []
            for j in range(self.cols):
                prod.append(
                    self.nf.GetProdV(
                    Label(lb + "_prod", i, j),
                    self.getweight(i, j), inputs[j]))

            sum.append(self.nf.BuildSum2VTree(prod, lb))

        # these help weight addition
        self.inlabels = map(lambda node: node.GetLabel(), inputs)
        self.outlabels = map(lambda node: node.GetLabel(), sum)

        return sum

class BiasedLinearMap(LinearMap):
    
    def __init__(self, net, indim, outdim, labelbase,
                 meanprior = None, varprior = None,
                 initvalues = None):
        LinearMap.__init__(self, net, indim, outdim, labelbase, initvalues)
        self.__addbias(meanprior, varprior)

    def __addbias(self, meanprior, varprior):
        lb = self.labelbase

        if meanprior is None:
            meanprior = self.net.GetNode("c0")
        if varprior is None:
            varprior = self.net.GetNode("c_5")

        for i in range(self.rows):
            b = self.nf.GetGaussian(self.bias_label(i), meanprior, varprior)
            b.SetPersist(5)

    def bias_label(self, i):
        return Label(self.labelbase + "_b", i)

    def getbias(self, i):
        return self.net.GetVariable(self.bias_label(i))

    def linmap(self, inputs):
        lb = self.labelbase
        sum = LinearMap.linmap(self, inputs)

        biased = []
        for i in range(self.rows):
            biased.append(self.nf.GetSum2V(
                Label(lb + "_bias", i), sum[i], self.getbias(i)))

        return biased

class GenericIca:
    def __init__(self, net, sdim, xdim, umpar, uvpar, labelbase,
                 linmap = LinearMap):
        self.net = net
        self.nf = PyNodeFactory(net)
        self.sdim = sdim
        self.xdim = xdim
        self.umpar = umpar
        self.uvpar = uvpar
        self.labelbase = labelbase
        self.linmap = linmap
        self.__buildblock()

    def createsources(self, vpar):
        raise "AbstractMethodInvocationError"

    def initsources(self, data, decay):
        raise "AbstractMethodInvocationError"

    def rotatesources(self, decay):
        raise "AbstractMethodInvocationError"

    def setsources(self, values, decay):
        sources = self.getsources()
        assert values.shape[0] == len(sources)
        for i in range(len(sources)):
            self.nf.EvidenceVNode(sources[i], mean=values[i,:], decay=decay)

    def source_label(self, i):
        return Label(self.labelbase + "_s", i)

    def getsource(self, i):
        return self.net.GetVariable(self.source_label(i))

    def getsources(self):
        """Returns sources that are still alive."""
        sources = []
        for i in range(self.sdim):
            s = self.getsource(i)
            if s is not None:
                sources.append(s)
        return sources

    def createvarneurons(self):
        return self.nf.MakeNodes(
            "GaussianV", self.labelbase + "_u", self.sdim,
            self.umpar, self.uvpar)

    def __buildblock(self):
        u = self.createvarneurons()
        s = self.createsources(u)
        smap = self.linmap(self.net, self.sdim, self.xdim,
                           self.labelbase + "_A")
        outputs = smap.linmap(s)

        map(lambda nodelst: map(lambda node: node.SetPersist(5), nodelst),
            [s, u])

        self.u = u
        self.s = s
        self.smap = smap
        self.outputs = outputs

class Ica(GenericIca):
    def createsources(self, vpar):
        c0 = self.net.GetNode("c0")
        s = []
        for i in range(self.sdim):
            s.append(self.nf.GetGaussianV(
                self.source_label(i), c0, vpar[i]))
        return s

    def initsources(self, data, decay):
        pc = Helpers.DoPCA(util.normalize(data), self.sdim)
        for i in range(self.sdim):
            self.nf.EvidenceVNode(self.s[i], mean=pc[i,:], decay=decay)

class DynIca(GenericIca):
    def createsources(self, vpar):
        c0, c1, c_5 = map(self.net.GetNode, ["c0", "c1", "c_5"])
        s = []
        for i in range(self.sdim):
            s.append(self.nf.GetDelayGaussV(
                self.source_label(i), c0, vpar[i], c1, c0, c_5))
        return s

    def initsources(self, data, decay):
        pca = Helpers.DoPCA(util.smooth(util.normalize(data), 20), self.sdim)
        GenericIca.setsources(self, pca, decay)

class ArIca(DynIca):
    def createsources(self, vpar):
        c0, c1, c_5 = map(self.net.GetNode, ["c0", "c1", "c_5"])
        lb = self.labelbase
        s = []
        for i in range(self.sdim):
            a = self.nf.GetGaussian(Label(lb + "_as", i), c0, c_5)
            s.append(self.nf.GetDelayGaussV(Label(lb + "_s", i),
                                 c0, vpar[i], a, c0, c_5))
        return s

class DynVar(DynIca):
    def createvarneurons(self):
        c0, c1, c_5 = map(self.net.GetNode, ["c0", "c1", "c_5"])
        lb = self.labelbase + "_u"
        u = []
        for i in range(self.sdim):
            u.append(self.nf.GetDelayGaussV(
                Label(lb, i), self.umpar[i], self.uvpar[i], c1, c0, c_5))
        return u