objecthmm.py

General Hidden Markov Model Library 一个通用的隐马尔科夫模型的C代码库

        cstate.b = ghmmhelper.list2double_array(self.weights)

    def ReadCState(self, cstate, M):
        self.weights = ghmmhelper.double_array2list(cstate.b, M)


class DiscreteHigherOrderEmission(DiscreteEmission):
    def __init__(self, alphabet, order=0):
        self.alphabet = alphabet
        self.order    = ValidatingInt(order)
        M = self.alphabet.size()
        if M >  0:
            self.weights = [1.0 / M] * M**(order+1)
        else:
            self.weights = []

    def grow(self):
        print "BUG! DiscreteHigherOrderEmission.grow() not implemented and should be called"
        
    def shrink(self, index):
        print "BUG! DiscreteHigherOrderEmission.shrink() not implemented and should be called"

    def edit(self, master, description):
        if self.order > 0:
            message = "editing the emissions of higher order states is not implemented"
            tkMessageBox.showwarning("HMMEd", message)
            print message
        else:
            DiscreteEmission.edit(self, master, description)

    def ChangeOrder(self, neworder):
        M = self.alphabet.size()
        if neworder < self.order:
            self.weights = self.weights[0:M**(neworder+1)]
        elif neworder > self.order:
            self.weights += [1.0 / M] * (M**(neworder+1) - M**neworder)
        self.order = neworder

    def ReadCState(self, cstate, M):
        self.order   = ValidatingInt(cstate.order)
        self.weights = ghmmhelper.double_array2list(cstate.b, M**(self.order+1))


class ContinuousEmission(Emission):
    def __init__(self):
        Emission.__init__(self)
        self.weights  = [1.0]
        self.plotList = []

    def get(self):
        raise

    def set(self, value):
        raise

    def edit(self, master, stateId):
        if len(self.plotList) == 0:
            self.plotList.append(NormalDensity(mu=0,sigma=1,color="LightGreen"))

        tmp = [copy.copy(x) for x in self.plotList]
        top = Tkinter.Toplevel(master)
        d = HMMEditor.ContinuousEmissionEditor(top, tmp)
        d.pack(expand=1,fill=Tkinter.BOTH)
        top.withdraw()
        top.title( ("State %d emissions" % stateId))        
        top.update_idletasks()
        top.deiconify()
        top.wait_window(top)
        
        if d.success():
            self.plotList = d.plot_list
            self.weights  = [ d.dict[str(i)] for i in xrange(1,len(self.plotList)+1)]

    def writeParameters(self, cstate):
        cstate.M = len(self.plotList)
        
        muL=[]; uL=[]; aL=[]; density_tL=[]; cL=[]
        for d in xrange(len(self.plotList)): 
            (mu, u, a, density_t) = self.plotList[d].getParameters()
            muL.append(mu); uL.append(u); aL.append(a); density_tL.append(density_t);

        # write parameters in the state
        density_array  = ghmmwrapper.density_array_alloc(len(density_tL))
        for (i,density) in enumerate(density_tL):
            ghmmwrapper.density_array_setitem(density_array, i, density)
        cstate.density = density_array

        cstate.mue = ghmmhelper.list2double_array(muL)
        cstate.u   = ghmmhelper.list2double_array(uL)
        cstate.a   = ghmmhelper.list2double_array(aL)
        cstate.c   = ghmmhelper.list2double_array(self.weights)

        # editor doesn't supports fixed mixture components
        cstate.mixture_fix = ghmmhelper.list2int_array([0] * len(self.plotList))

    def ReadCState(self, cstate, M):
        M = cstate.M
        for i in xrange(M):
            if cstate.getDensity(i) == ghmmwrapper.normal:
                self.plotList.append(NormalDensity())
            elif cstate.getDensity(i) == ghmmwrapper.normal_left:
                self.plotList.append(NormalDensityTruncLeft())
            elif cstate.getDensity(i) == ghmmwrapper.normal_right:
                self.plotList.append(NormalDensityTruncRight())
            elif cstate.getDensity(i) == ghmmwrapper.uniform:
                self.plotList.append(UniformDensity())



class Distribution(ContinuousEmission):
    def __str__(self):
        return "  1.0  *  " + str(self.get())

class UniformDistribution(Distribution):
    def __init__(self, start=0, stop=1):
        ContinuousEmission.__init__(self)
        self.plotList = [UniformDensity(start, stop)]

    def set(self, values):
        self.weights = [1.0]
        density = self.plotList[0]
        density.start = float(min(values))
        density.stop  = float(max(values))

    def get(self):
        return self.plotList[0]

    def getParameters(self):
        return self.plotList[0].getParameters()

class GaussianDistribution(Distribution):
    def __init__(self, mu=0, sigma=1):
        ContinuousEmission.__init__(self)
        self.plotList = [NormalDensity(mu, sigma)]

    def set(self, values):
        self.weights = [1.0]
        density = self.plotList[0]
        density.mu    = float(values[0])
        density.sigma = float(values[1])

    def get(self):
        return self.plotList[0]

    def getParameters(self):
        return self.plotList[0].getParameters()

class LeftTruncGaussianDistribution(Distribution):
    def __init__(self, mu=0, sigma=1, trunc=0.5):
        ContinuousEmission.__init__(self)
        self.plotList = [NormalDensityTruncLeft(mu, sigma, trunc)]

    def set(self, values):
        self.weights = [1.0]
        density = self.plotList[0]
        density.mu    = float(values[0])
        density.sigma = float(values[1])
        density.tail  = float(values[2])

    def get(self):
        return self.plotList[0]

    def getParameters(self):
        return self.plotList[0].getParameters()

class RightTruncGaussianDistribution(Distribution):
    def __init__(self, mu=0, sigma=1, trunc=0.5):
        ContinuousEmission.__init__(self)
        self.plotList = [NormalDensityTruncRight(mu, sigma, trunc)]

    def set(self, values):
        self.weights = [1.0]
        density = self.plotList[0]
        density.mu    = float(values[0])
        density.sigma = float(values[1])
        density.tail  = float(values[2])

    def get(self):
        return self.plotList[0]

    def getParameters(self):
        return self.plotList[0].getParameters()

class ContinuousMixtureDistribution(Distribution):
    def __init__(self, parameters=[]):
        ContinuousEmission.__init__(self)
        self.set(parameters)

    def __str__(self):
        string = ""
        for i,w in enumerate(self.weights):
            string += ("  %f  *  " % w) + str(self.plotList[i]) + '\n'
        return string[:-1]

    def set(self, values):
        self.weights = []
        self.plotList = []
        if isinstance(values, Distribution):
            values = [(values, 1.0)]
        for value in values:
            self.plotList.append(value[0])
            if len(value) > 1:
                self.weights.append(value[1])
            else:
                self.weights.append(-1)

        weights = [w for w in self.weights if w >= 0.0]
        wsum = sum(weights)
        if wsum > 1.0:
            factor = 1.0 / wsum
            for i,weight in enumerate(self.weights):
                if weight >= 0.0:
                    self.weights[i] *= factor
            wsum = 1.0
        if len(weights) < len(self.weights):
            mean = (1.0-wsum) / (len(self.weights)-len(weights))
            for i,weight in  enumerate(self.weights):
                if weight >= 0.0:
                    self.weights[i] = mean
            

    def get(self):
        retlist = []
        for (i, w) in enumerate(self.weights):
            retlist.append((self.plotList[i], w))
        return retlist

class GaussianMixtureDistribution(ContinuousMixtureDistribution):
    pass


class DiscretePairEmission(Emission):
    pass


class State(VertexObject):

    def __init__(self, emission=Emission(), hmm=None):
        VertexObject.__init__(self)
        self.num = -1   # continuous id (0..Order()-1) used for writing transitions
        self.editableAttr = {'labeling':"Name", 'initial':"Initial Probability", 'fixed':"fixed emissions"}
        self.initial = Probability()
        self.Emission = emission
        self.itsHMM = hmm
        self.fixed = ValidatingBool(False)

    def __setattr__(self, name, value):
        if name is "emission":
            self.Emission.set(value)
        elif name is "name":
            try:
                if self.itsHMM is not None:
                    vname = self.labeling
                    self.itsHMM.name2id[vname] = [x for x in self.itsHMM.name2id[vname] if x != self.id]
                    if len(self.itsHMM.name2id[vname]) == 0:
                        del self.itsHMM.name2id[vname]
            except:
                raise "State doesn't hold pointer to its HMM, can't update the name"
            self.labeling = ValidatingString(str(value))
        else:
            self.__dict__[name] = value
            #object.__setattr__(self, name, value)

    def __getattr__(self, name):
        if name is "emission":
            return self.Emission.get()
        elif name is "name":
            return self.labeling
        else:
            return self.__dict__[name]
            #return object.__getattr__(self, name)

    def __str__(self):
        string  = ('State %d ("%s"). Initial probability: %f\n' % (self.id, self.labeling, self.initial))
        string += str(self.Emission)
        return string
        
    def update(self):
        pass

    def normalize(self):
        # normalize outgoing transmission probabilities
        weights = [e.GetWeight() for e in self.outEdges]
        weights = [w for w in weights if w >= 0.0]
        osum = float(sum(weights))
        if osum > 1.0 or len(weights) == len(self.outEdges):
            for i,edge in enumerate(self.outEdges):
                w = edge.GetWeight()
                if w >= 0.0:
                    edge.SetWeight(w/osum)
            osum=1.0
        if len(weights) < len(self.outEdges):
            mean = (1.0-osum) / (len(self.outEdges)-len(weights))
            for i,edge in enumerate(self.outEdges):
                if edge.GetWeight() < 0.0:  
                    edge.SetWeight(mean)
            

    def editProperties(self, parent, attributes = None):
        self.update()
        self.desc = ('Properties of State %d (%s)' % (self.id, self.labeling))
        if self.itsHMM.modelType & ghmmwrapper.kHigherOrderEmissions:
            self.order = self.Emission.order
            self.editableAttr['order'] = "Order"
        if attributes == None:
            editBox = EditObjectAttributesDialog(parent, self, self.editableAttr)
        else:
            editableAttr = {}
            for attr in attributes:
                editableAttr[attr] = self.editableAttr[attr]
            editBox = EditObjectAttributesDialog(parent, self, editableAttr)
        if self.itsHMM.modelType & ghmmwrapper.kHigherOrderEmissions:
            self.Emission.ChangeOrder(self.order)
            del self.order
            del self.editableAttr['order']

    def editEmissions(self, master):
        self.Emission.edit(master, ('Properties of State %d (%s)' % (self.id, self.labeling)))

    def WriteCState(self, cstate):
        cstate.pi         = self.initial
        cstate.in_states  = len(self.inEdges)
        cstate.out_states = len(self.outEdges)

        cstate.desc      = str(self.labeling)
        if self.embedding is not None:
            cstate.xPosition = int(self.embedding.x)
            cstate.yPosition = int(self.embedding.y)
        else:
            cstate.xPosition = 0
            cstate.yPosition = 0

        cstate.fix = self.fixed

        self.WriteTransitions(cstate)
        self.Emission.writeParameters(cstate)

    def WriteTransitions(self, cstate):
        inID = [edge.tail.num for edge in self.inEdges]
        inA  = [edge.GetEdgeWeight(0) for edge in self.inEdges]
        cstate.in_id = ghmmhelper.list2int_array(inID)
        cstate.in_a  = ghmmhelper.list2double_array(inA)

        outID = [edge.head.num for edge in self.outEdges]
        outA = [edge.GetEdgeWeight(0) for edge in self.outEdges]
        cstate.out_id = ghmmhelper.list2int_array(outID)
        cstate.out_a  = ghmmhelper.list2double_array(outA)

    def ReadCState(self, cmodel, cstate, i):
        self.initial  = ValidatingFloat(cstate.pi)
        self.fixed    = ValidatingBool(cstate.fix)
        self.labeling = ValidatingString(cstate.desc)

        if self.itsHMM is not None:
            self.itsHMM.SetEmbedding(self.id, cstate.xPosition, cstate.yPosition)

        self.Emission.ReadCState(cstate, cmodel.M)


class ContinuousState(State):

    def WriteTransitions(self, cstate):
        inID = [edge.tail.id for edge in self.inEdges]
        inA  = [[edge.GetEdgeWeight(0) for edge in self.inEdges]]
        cstate.in_id = ghmmhelper.list2int_array(inID)
        (mat, lens)  = ghmmhelper.list2double_matrix(inA)
        cstate.in_a  = mat
        
        outID = [edge.head.id for edge in self.outEdges]
        outA  = [[edge.GetEdgeWeight(0) for edge in self.outEdges]]
        cstate.out_id = ghmmhelper.list2int_array(outID)
        (mat, lens)   = ghmmhelper.list2double_matrix(outA)
        cstate.out_a  = mat


class SilentState(State):
    def __init__(self, emission=Emission(), hmm=None, silent=ValidatingBool(False)):
        State.__init__(self, emission, hmm)
        self.init(silent)

    def init(self, silent):
        self.editableAttr['silent'] = "Silent"
        self.silent = silent

    def editEmissions(self, master):
        if not self.silent:
            State.editEmissions(self, master)