HMM的维特比算法简单示例

Posted 2020-08-28

今天读了一位大牛的关于HMM的技术博客，读完之后，写了一个关于维特比算法的简单示例，用scala和java语言混合编写的。现在上传之。

package com.txq.hmm import java.util
import scala.collection._ 
/**
 * HMM维特比算法，根据显示状态链条估计隐式链条
 * @param states 隐式states
 * @param observations 显式states   
 * @param start_probability 初始概率向量   
 * @param transition_probability 转换概率   
 * @param emission_probability 发射概率   
 * @author TongXueQiang   
 * @ 2017/01/21   
 */ 
class Markov_weather(states:Array[Any],observations:Array[Any],start_probability:Map[String,Double],transition_probability:Map[String,Map[String,Double]],emission_probability:Map[String,Map[String,Double]]) {
   /**
    * 
    * @param obs_ser 观察到的显式状态链条     
    * @return     
    */   
   def compute_hidden_ser(obs_ser:util.LinkedList[String]):util.LinkedList[String] = {
     val result = new util.LinkedList[String]()//把结果放到链表中 
     var p = mutable.Map[String,Double]()//存储转换概率和输出概率的乘积
     var keys:Seq[String] = null
     var values:Seq[Double] = null 
     var max:Double = 0.0
     var label:Int = 0
     //初始化,先得到一个隐式的状态
     var obs_state = obs_ser.poll()
     if(obs_state != null){
       for((k,v) <- start_probability){ 
        val a = v * emission_probability(k)(obs_state)
         p += (k->a)
       }
       keys = p.keys.toSeq
       values = p.values.toSeq
       max = values(0)
       for(i <- 1 until values.size){
         if(values(i) > max){
           max = values(i)
           label = i
         }
       }
       result.add(keys(label))
       label = 0
       p.clear()
     }
     //开始迭代
     obs_state = obs_ser.poll()
     while(obs_state != null){
       val hid_ser = result.peekLast()
       if(hid_ser != null){
         for((k,v) <- transition_probability(hid_ser)){
           val a = v * emission_probability(k)(obs_state)
           p += (k->a)
         }
         keys = p.keys.toSeq
         values = p.values.toSeq
         max = values(0)
         for(i <- 1 until values.size){
           if(values(i) > max){
             max = values(i)
             label = i
           }
         }
         result.add(keys(label)) 
        label = 0 
        p.clear()
        obs_state = obs_ser.poll()
       }
     }
     result
   }
 }

 测试类：

package com.txq.hmm import java.util 
/** 
  *
  */
 object HmmTest {
   def main(args:Array[String]):Unit = {
     //测试数据集
     val states = Array[Any]("Rainy","Sunny")//隐式states 
     val observations = Array[Any]("walk","shop","clean")//显式states
     val start_probability = Map("Rainy"->0.6,"Sunny"->0.4)//初始概率向量
     val transition_probability = Map("Rainy"->Map("Rainy"->0.7,"Sunny"->0.3),"Sunny"->Map("Rainy"->0.4,"Sunny"->0.6))//转换概率
     //发射概率
     val emission_probability = Map("Rainy"->Map("walk"->0.1,"shop"->0.4,"clean"->0.5),"Sunny"->Map("walk"->0.6,"shop"->0.3,"clean"->0.1))
     val obs_ser = new util.LinkedList[String]()//观察到的显式状态集
     obs_ser.add("walk")
     obs_ser.add("shop")
     obs_ser.add("clean")
     val markov:Markov_weather = new Markov_weather(states,observations,start_probability,transition_probability,emission_probability)
     val result = markov.compute_hidden_ser(obs_ser)//执行计算
     var hidden_state = result.pollFirst()
     while(hidden_state != null){
       print(hidden_state)
       if(result.size() != 0){
         print("->")
       }
       hidden_state = result.pollFirst()
     } 
  }
 }
显示的结果：
Sunny->Sunny->Rainy
Process finished with exit code 0