objecthmm.py

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

#!/usr/bin/env python
################################################################################
#
#       This file is part of Gato (Graph Algorithm Toolbox) 
#
#	file:   ObjectHMM.py
#	author: Janne Grunau
#
#       Copyright (C) 1998-2002, Alexander Schliep
#                                   
#       Contact: [e-mail]
#
#       Information: http://gato.sf.net
#
#       This library is free software; you can redistribute it and/or
#       modify it under the terms of the GNU Library General Public
#       License as published by the Free Software Foundation; either
#       version 2 of the License, or (at your option) any later version.
#
#       This library 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
#       Library General Public License for more details.
#
#       You should have received a copy of the GNU Library General Public
#       License along with this library; if not, write to the Free
#       Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
#
#
#       This file is version $Revision: 1.29 $ 
#                       from $Date: 2005/02/22 11:12:56 $
#             last change by $Author: schliep $.
#
################################################################################

from Gato.ObjectGraph import *
from Gato.EditObjectAttributesDialog import *
from Gato.MapEditor import NamedCollectionEditor
from Gato import ProbEditorBasics, ProbEditorDialogs, ProbEditorContinuous

import Tkinter

import ghmmwrapper, ghmmhelper, ghmm, HMMEditor

import copy

try:
    from collections import defaultdict
except:
    class defaultdict(dict):
        def __init__(self, default_factory=None, *a, **kw):
            if (default_factory is not None and
                not hasattr(default_factory, '__call__')):
                raise TypeError('first argument must be callable')
            dict.__init__(self, *a, **kw)
            self.default_factory = default_factory
        def __getitem__(self, key):
            try:
                return dict.__getitem__(self, key)
            except KeyError:
                return self.__missing__(key)
        def __missing__(self, key):
            if self.default_factory is None:
                raise KeyError(key)
            self[key] = value = self.default_factory()
            return value
        def __reduce__(self):
            if self.default_factory is None:
                args = tuple()
            else:
                args = self.default_factory,
            return type(self), args, None, None, self.items()
        def copy(self):
            return self.__copy__()
        def __copy__(self):
            return type(self)(self.default_factory, self)
        def __deepcopy__(self, memo):
            import copy
            return type(self)(self.default_factory,
                              copy.deepcopy(self.items()))
        def __repr__(self):
            return ('defaultdict(%s, %s)' % (self.default_factory,
                                             dict.__repr__(self)))


class DiscreteHMMAlphabet:
    def __init__(self, names = [], description = "alphabet_1"):
        self.id = description
        self.name = {}
        self.name2code = {}
        for i,name in enumerate(names):
            self.name[i]         = name
            self.name2code[name] = i

    def __str__(self):
        string  = str(self.id) + ": "
        string += str(self.name)
        return string

    def size(self):
        return len(self.name.keys())

    def edit(self, master, name):
        pass

    def editDialog(self, master, hmm):
        self.hmm = hmm
        editor = NamedCollectionEditor(master, self)
        self.hmm = None

    def names(self):
        return self.name.values()

    def add(self, name):
        key = len(self.name)
        self.name[key] = name
        self.name2code[name] = key
        # update all states emissions
        for state in self.hmm.vertices.values():
            state.emission.grow()
        # also update background distributions if needed
        if self.hmm.modelType & ghmmwrapper.kBackgroundDistributions:
            self.hmm.backgroundDistributions.grow()

    def delete(self, name):
        key = self.name2code[name]
        for i in range(key, len(self.name)-1):
            self.name[i] = self.name[i+1]
            self.name2code[self.name[i]] = i

        del self.name2code[name]
        del self.name[len(self.name)-1]

        # update all states emissions
        for state in self.hmm.vertices.values():
            state.emission.shrink(key)
        # also update background distributions if needed
        if self.hmm.modelType & ghmmwrapper.kBackgroundDistributions:
            self.hmm.backgroundDistributions.shrink()
        # and tie groups
        if self.hmm.modelType & ghmmwrapper.kTiedEmissions:
            self.hmm.tie_groups.shrink()

    def GetKeys(self):
        return self.name.keys()

    def GetKey(self, symbol):
        return self.name2code[symbol]
    
    def GetSymbol(self, key):
        return self.name[key]

    def ReadCAlphabet(self, calphabet):
        self.id = calphabet.description
        if self.id is None:
            self.id = str(calphabet.size)
        for i in xrange(0, calphabet.size):
            name = calphabet.getSymbol(i)
            self.name[i] = name
            self.name2code[name] = i

    def WriteCAlphabet(self):
        calphabet = ghmmwrapper.ghmm_alphabet(len(self.name), self.id )
        #calphabet.description = self.id 
        for i in xrange(len(self.name)):
            calphabet.setSymbol(i, self.name[i])

        return calphabet


class DiscreteHMMBackground:
    def __init__(self, eclass):
        self.EmissionClass = eclass
        self.nextKey = 1
        self.val2pop = {0:"no background"}
        self.name = {}
        self.name2code = {}
        self.values = {}
        self.hmm = None

    def size(self):
        return len(self.name.keys())

    def edit(self, master, name):
        self.values[self.name2code[name]].edit(master, "backgound distribution \"%s\""%name)

    def editDialog(self, master, hmm):
        self.hmm = hmm
        editor = NamedCollectionEditor(master, self)
        self.hmm = None

    def names(self):
        return self.name.values()

    def add(self, name, alphabet=None):
        key = self.nextKey
        self.nextKey += 1
        
        if self.hmm is not None:
            e = self.EmissionClass(self.hmm.alphabet)
        elif alphabet is not None:
            e = self.EmissionClass(alphabet)
        else:
            e = self.EmissionClass()

        self.name[key] = name
        self.name2code[name] = key
        self.values[key] = e
        self.val2pop[key] = name

    def delete(self, name):
        key = self.name2code[name]
        del self.name[key]
        del self.name2code[name]
        del self.values[key]
        del self.val2pop[key]

    def grow(self):
        for emission in self.values.values():
            emission.grow()

    def shrink(self):
        for emission in self.values.values():
            emission.shrink()

    def getOrders(self):
        keys = self.name.keys()
        keys.sort()
        return [self.values[k].order for k in keys]

    def getWeights(self):
        keys = self.name.keys()
        keys.sort()
        return [self.values[k].weights for k in keys]

    def getNames(self):
        keys = self.name.keys()
        keys.sort()
        return [self.name[k] for k in keys]

    def ReadCBackground(self, alphabet, bp):
        number  = bp.n
        M       = bp.m
        for i in xrange(number):
            key = self.nextKey
            self.nextKey += 1

            e = self.EmissionClass(alphabet)
            e.order   = bp.getOrder(i)
            e.weights = ghmmhelper.double_array2list(bp.getWeights(i), M**(e.order+1))

            name = bp.getName(i)

            self.name[key] = name
            self.name2code[name] = key
            self.values[key] = e
            self.val2pop[key] = name

class TieGroups(DiscreteHMMBackground):

    def __init__(self, eclass):
        self.EmissionClass = eclass
        self.nextKey = 1
        self.val2pop = {0:"untied"}
        self.name = {}
        self.name2code = {}
        self.values = {}
        self.hmm = None

    def edit(self, master, name):
        self.values[self.name2code[name]].edit(master, "Emissions of tie group \"%s\""%name)

    def editEmissions(self, master, id):
        self.values[id].edit(master, "Emissions of tie group \"%s\""%self.name[id])

    def ReadCBackground(self, alphabet, bp):
        pass


class UniformDensity(ProbEditorContinuous.box_function):
    def getParameters(self):
        return (self.stop, self.start, 0, ghmmwrapper.uniform)

class NormalDensity(ProbEditorContinuous.gauss_function):
    def getParameters(self):
        return (self.mu, self.sigma, 0, ghmmwrapper.normal)

class NormalDensityTruncRight(ProbEditorContinuous.gauss_tail_function_right):
    def getParameters(self):
        return (self.mu, self.sigma, self.tail, ghmmwrapper.normal_right)
    
class NormalDensityTruncLeft(ProbEditorContinuous.gauss_tail_function_left):
    def getParameters(self):
        return (self.mu, self.sigma, self.tail, ghmmwrapper.normal_left)

class Emission(object):
    def __init__(self):
        pass

    def __str__(self):
        pass

    def edit(self, master):
        pass

    def get(self):
        pass
    
    def set(self, value):
        pass

    def writeParameters(self, cstate):
        pass

    def ReadCState(self, cstate, M):
        pass

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

    def __str__(self):
        return str(self.weights)

    def grow(self):
        s = float(self.alphabet.size()-1)
        self.weights = [(x*s) for x in self.weights] + [1.0]
        self.weights = [x/sum(self.weights) for x in self.weights]
        
    def shrink(self, index):
        del self.weights[index]
        s = sum(self.weights)
        if s > 0.0:
            self.weights = [x / s for x in self.weights]
        elif self.alphabet.size() > 0:
            self.weights = [1.0 / self.alphabet.size()] * self.alphabet.size()
        else:
            self.weights = []

    def edit(self, master, description):
        transition_probabilities = ProbEditorBasics.ProbDict({})
        for key in self.alphabet.GetKeys():
            weight = self.weights[key]
            label  = self.alphabet.GetSymbol(key)
            transition_probabilities.update({label:weight})
            
        if transition_probabilities.sum == 0:
            key_list = transition_probabilities.keys()
            for key in key_list:
                transition_probabilities[key]=1.0/len(key_list)
        e = ProbEditorBasics.emission_data(transition_probabilities)
        d = ProbEditorDialogs.emission_dialog(master, e, description)

        if d.success():
            # write back normalized probabilities
            for label in transition_probabilities.keys():
                key = self.alphabet.GetKey(label)
                self.weights[key] = transition_probabilities[label]/transition_probabilities.sum

    def get(self):
        return self.weights

    def set(self, values):
        diff = self.alphabet.size() - len(values)
        #print diff, values, self.alphabet.size(), self.alphabet
        if diff == 0:
            self.weights = values
        elif diff > 0:
            s = sum(values)
            if s == 1.0:
                values += ([0.0]*diff)
            elif s < 1.0 and s >= 0.0:
                values += ([(1.0-s)/diff] * diff)
            else:
                raise ValueError("sum")
        else:
            raise ValueError("wrong number of arguments")

        self.weights = values

    def writeParameters(self, cstate):