超详细匈⽛利算法流程以及Python程序实现通俗易懂前不久在⽆⼈机检测跟踪的项⽬中⽤到了多⽬标跟踪算法（该项⽬后续会发贴介绍），其中需要涉及多个⽬标在两帧之间的匹配问题，最初使⽤的是最简单的距离最⼩化原则进⾏帧间多⽬标的匹配。
后来通过实习和查阅论⽂等渠道了解到了多⽬标跟踪领域经典的Sort和DeepSort算法，其中都使⽤到了匈⽛利算法解决匹配问题，因此开此贴记录⼀下算法的学习过程。
指派问题概述⾸先，对匈⽛利算法解决的问题进⾏概述：实际中，会遇到这样的问题，有n项不同的任务，需要n个⼈分别完成其中的1项，每个⼈完成任务的时间不⼀样。
于是就有⼀个问题，如何分配任务使得花费时间最少。
通俗来讲，就是n*n矩阵中，选取n个元素，每⾏每列各有1个元素，使得和最⼩。
如下表所⽰上表可以抽象成⼀个矩阵，如果是如上表所⽰的求和最⼩问题，那么这个矩阵就叫做花费矩阵（Cost Matrix）；如果要求的问题是使之和最⼤化，那么这个矩阵就叫做利益矩阵（Profit Matrix）。
匈⽛利算法流程算法流程匈⽛利算法有多种实现⽅式，⽐如基于图论的⽅式等，本⽂主要使⽤矩阵变换来实现，这种⽅式你甚⾄可以在纸上写写画画，⽽且理解起来⽐较简单。
本⽂算法流程如上图所⽰，⾸先进⾏列规约，即每⾏减去此⾏最⼩元素，每⼀列减去该列最⼩元素，规约后每⾏每列中必有0元素出现。
接下来进⾏试指派，也就是划最少的线覆盖矩阵中全部的0元素，如果试指派的独⽴0元素数等于⽅阵维度则算法结束，如果不等于则需要对矩阵进⾏调整，重复试指派和调整步骤直到满⾜算法结束条件。
以上是我简要描述的算法流程，值得⼀提的是，⽤矩阵变换求解的匈⽛利算法也有多种实现，主要不同就在于试指派和调整矩阵这块，但万变不离其宗都是为了⽤最少的线覆盖矩阵中全部的零元素。
咱们废话少说，来看⼀个例⼦。
程序实现完整代码（带测试⽤例）'''@Date: 2020/2/23@Author:ZhuJunHui@Brief: Hungarian Algorithm using Python and NumPy'''import numpy as npimport collectionsimport timeclass Hungarian():""""""def__init__(self, input_matrix=None, is_profit_matrix=False):"""输⼊为⼀个⼆维嵌套列表is_profit_matrix=False代表输⼊是消费矩阵（需要使消费最⼩化），反之则为利益矩阵（需要使利益最⼤化） """if input_matrix is not None:# 保存输⼊my_matrix = np.array(input_matrix)self._input_matrix = np.array(input_matrix)self._maxColumn = my_matrix.shape[1]self._maxRow = my_matrix.shape[0]# 本算法必须作⽤于⽅阵，如果不为⽅阵则填充0变为⽅阵matrix_size =max(self._maxColumn, self._maxRow)pad_columns = matrix_size - self._maxRowpad_rows = matrix_size - self._maxColumnmy_matrix = np.pad(my_matrix,((0,pad_columns),(0,pad_rows)),'constant', constant_values=(0)) # 如果需要，则转化为消费矩阵if is_profit_matrix:my_matrix = self.make_cost_matrix(my_matrix)self._cost_matrix = my_matrixself._size =len(my_matrix)self._shape = my_matrix.shape# 存放算法结果self._results =[]self._totalPotential =0else:self._cost_matrix =Nonedef make_cost_matrix(self,profit_matrix):'''利益矩阵转化为消费矩阵，输出为numpy矩阵'''# 消费矩阵 = 利益矩阵最⼤值组成的矩阵 - 利益矩阵matrix_shape = profit_matrix.shapeoffset_matrix = np.ones(matrix_shape, dtype=int)* profit_matrix.max()cost_matrix = offset_matrix - profit_matrixreturn cost_matrixdef get_results(self):"""获取算法结果"""return self._resultsdef calculate(self):"""实施匈⽛利算法的函数"""result_matrix = self._cost_matrix.copy()# 步骤 1: 矩阵每⼀⾏减去本⾏的最⼩值for index, row in enumerate(result_matrix):result_matrix[index]-= row.min()# 步骤 2: 矩阵每⼀列减去本⾏的最⼩值for index, column in enumerate(result_matrix.T):result_matrix[:, index]-= column.min()#print('步骤2结果 ',result_matrix)#print('步骤2结果 ',result_matrix)# 步骤 3：使⽤最少数量的划线覆盖矩阵中所有的0元素# 如果划线总数不等于矩阵的维度需要进⾏矩阵调整并重复循环此步骤total_covered =0while total_covered < self._size:time.sleep(1)#print("---------------------------------------")#print('total_covered: ',total_covered)#print('result_matrix:',result_matrix)# 使⽤最少数量的划线覆盖矩阵中所有的0元素同时记录划线数量cover_zeros = CoverZeros(result_matrix)single_zero_pos_list = cover_zeros.calculate()covered_rows = cover_zeros.get_covered_rows()covered_columns = cover_zeros.get_covered_columns()total_covered =len(covered_rows)+len(covered_columns)# 如果划线总数不等于矩阵的维度需要进⾏矩阵调整（需要使⽤未覆盖处的最⼩元素）if total_covered < self._size:result_matrix = self._adjust_matrix_by_min_uncovered_num(result_matrix, covered_rows, covered_columns) #元组形式结果对存放到列表self._results = single_zero_pos_list# 计算总期望结果value =0for row, column in single_zero_pos_list:value += self._input_matrix[row, column]self._totalPotential = valuedef get_total_potential(self):return self._totalPotentialdef_adjust_matrix_by_min_uncovered_num(self, result_matrix, covered_rows, covered_columns):"""计算未被覆盖元素中的最⼩值（m）,未被覆盖元素减去最⼩值m,⾏列划线交叉处加上最⼩值m"""adjusted_matrix = result_matrix# 计算未被覆盖元素中的最⼩值（m）elements =[]for row_index, row in enumerate(result_matrix):if row_index not in covered_rows:for index, element in enumerate(row):if index not in covered_columns:elements.append(element)min_uncovered_num =min(elements)#print('min_uncovered_num:',min_uncovered_num)#未被覆盖元素减去最⼩值mfor row_index, row in enumerate(result_matrix):if row_index not in covered_rows:for index, element in enumerate(row):if index not in covered_columns:adjusted_matrix[row_index,index]-= min_uncovered_num#print('未被覆盖元素减去最⼩值m',adjusted_matrix)#⾏列划线交叉处加上最⼩值mfor row_ in covered_rows:for col_ in covered_columns:#print((row_,col_))adjusted_matrix[row_,col_]+= min_uncovered_num#print('⾏列划线交叉处加上最⼩值m',adjusted_matrix)return adjusted_matrixclass CoverZeros():"""使⽤最少数量的划线覆盖矩阵中的所有零输⼊为numpy⽅阵""""""def__init__(self, matrix):# 找到矩阵中零的位置（输出为同维度⼆值矩阵，0位置为true，⾮0位置为false）self._zero_locations =(matrix ==0)self._zero_locations_copy = self._zero_locations.copy()self._shape = matrix.shape# 存储划线盖住的⾏和列self._covered_rows =[]self._covered_columns =[]def get_covered_rows(self):"""返回覆盖⾏索引列表"""return self._covered_rowsdef get_covered_columns(self):"""返回覆盖列索引列表"""return self._covered_columnsdef row_scan(self,marked_zeros):'''扫描矩阵每⼀⾏，找到含0元素最少的⾏，对任意0元素标记（独⽴零元素），划去标记0元素（独⽴零元素）所在⾏和列存在的0元素''' min_row_zero_nums =[9999999,-1]for index, row in enumerate(self._zero_locations_copy):#index为⾏号row_zero_nums = collections.Counter(row)[True]if row_zero_nums < min_row_zero_nums[0]and row_zero_nums!=0:#找最少0元素的⾏min_row_zero_nums =[row_zero_nums,index]#最少0元素的⾏row_min = self._zero_locations_copy[min_row_zero_nums[1],:]#找到此⾏中任意⼀个0元素的索引位置即可row_indices,= np.where(row_min)#标记该0元素#print('row_min',row_min)marked_zeros.append((min_row_zero_nums[1],row_indices[0]))#划去该0元素所在⾏和列存在的0元素#因为被覆盖，所以把⼆值矩阵_zero_locations中相应的⾏列全部置为falseself._zero_locations_copy[:,row_indices[0]]= np.array([False for _ in range(self._shape[0])])self._zero_locations_copy[min_row_zero_nums[1],:]= np.array([False for _ in range(self._shape[0])])def calculate(self):'''进⾏计算'''#储存勾选的⾏和列ticked_row =[]ticked_col =[]marked_zeros =[]#1、试指派并标记独⽴零元素while True:#print('_zero_locations_copy',self._zero_locations_copy)#循环直到所有零元素被处理（_zero_locations中没有true）if True not in self._zero_locations_copy:breakself.row_scan(marked_zeros)#2、⽆被标记0（独⽴零元素）的⾏打勾independent_zero_row_list =[pos[0]for pos in marked_zeros]ticked_row =list(set(range(self._shape[0]))-set(independent_zero_row_list))#重复3,4直到不能再打勾TICK_FLAG =Truewhile TICK_FLAG:#print('ticked_row:',ticked_row,' ticked_col:',ticked_col)TICK_FLAG =False#3、对打勾的⾏中所含0元素的列打勾for row in ticked_row:#找到此⾏row_array = self._zero_locations[row,:]#找到此⾏中0元素的索引位置#找到此⾏中0元素的索引位置for i in range(len(row_array)):if row_array[i]==True and i not in ticked_col:ticked_col.append(i)TICK_FLAG =True#4、对打勾的列中所含独⽴0元素的⾏打勾for row,col in marked_zeros:if col in ticked_col and row not in ticked_row:ticked_row.append(row)FLAG =True#对打勾的列和没有打勾的⾏画画线self._covered_rows =list(set(range(self._shape[0]))-set(ticked_row)) self._covered_columns = ticked_colreturn marked_zerosif __name__ =='__main__':#以下为3个测试⽤例cost_matrix =[[4,2,8],[4,3,7],[3,1,6]]hungarian = Hungarian(cost_matrix)print('calculating...')hungarian.calculate()print("Expected value:\t\t12")print("Calculated value:\t", hungarian.get_total_potential())# = 12print("Expected results:\n\t[(0, 1), (1, 0), (2, 2)]")print("Results:\n\t", hungarian.get_results())print("-"*80)profit_matrix =[[62,75,80,93,95,97],[75,80,82,85,71,97],[80,75,81,98,90,97],[78,82,84,80,50,98],[90,85,85,80,85,99],[65,75,80,75,68,96]]hungarian = Hungarian(profit_matrix, is_profit_matrix=True)hungarian.calculate()print("Expected value:\t\t543")print("Calculated value:\t", hungarian.get_total_potential())# = 543print("Expected results:\n\t[(0, 4), (2, 3), (5, 5), (4, 0), (1, 1), (3, 2)]")print("Results:\n\t", hungarian.get_results())print("-"*80)profit_matrix =[[62,75,80,93,0,97],[75,0,82,85,71,97],[80,75,81,0,90,97],[78,82,0,80,50,98],[0,85,85,80,85,99],[65,75,80,75,68,0]]hungarian = Hungarian(profit_matrix, is_profit_matrix=True)hungarian.calculate()print("Expected value:\t\t523")print("Calculated value:\t", hungarian.get_total_potential())# = 523print("Expected results:\n\t[(0, 3), (2, 4), (3, 0), (5, 2), (1, 5), (4, 1)]")print("Results:\n\t", hungarian.get_results())print("-"*80)print("-"*80)总结如开篇所⾔，匈⽛利算法具有多种实现⽅式，可见该算法多么优秀，本⽂的实现⽅式不⼀定是最优的，但相对⽽⾔⽐较通俗易懂。
