使用改良版多值覆盖Dancing link X （舞蹈链）求解aquarium游戏
在上一篇文章中，我们通过改造了dancing link代码解出了aquarium游戏，并输出了正确答案。
但是之前的代码感觉有些慢，10*10的谜面都要跑24秒，而且感觉之前的dancing link代码有些不完善（存在重复查询问题）。这一篇文章介绍如何改良多值覆盖dancing link模板代码，还有如何在整体上优化这个游戏的解题流程。
之前的代码是从所有列中选择可能性最少的列进行突破，以减少查询宽度；但是在查询过程中发现了问题：之前查询过的较高占据值的行可能会再次被查询到，从而浪费不少时间。这里就需要在每个列的查询过程做额外处理：先从值最大的行开始消除，并且退出对这个列的遍历之前都不还原这些行。
之前遍历序列可能是这样：[1,2,3] [1,2,4] [1,3,2] [1,3,4] [1,4,2] [1,4,3]，现在的序列则是：[1,2,3] [1,2,4] [1,3,4] [2,3,4]，查询次数少了一些（别小看这少掉的2次·，反应到查询树里面影响可能显著，特别是最初的几层）
还有一些重要的优化：如果消除的行过多，导致最终能选的行总值无法满足要求，也要及时返回。为了随时随地查验总值，防止过度消除，在headerCell里面添加一个属性：xum_limit_num，并根据这个思路进行优化：
""" Implementation of Donald Knuth's Dancing Links Sparse Matrix as a circular doubly linked list. (http://arxiv.org/abs/cs/0011047) """ import random import numpy as np __author__ = "FunCfans" class CannotAddRowsError(Exception): pass # class EmptyDLMatrix(Exception): pass # class Cell: """ Inner cell, storing 4 pointers to neighbors, a pointer to the column header and the indexes associated. """ __slots__ = list("UDLRC") + ["indexes", "limit_num"] def __init__(self, limitNum=1): self.U = self.D = self.L = self.R = self self.C = None self.indexes = None self.limit_num = limitNum def __str__(self): return f"Node: {self.indexes}" def __repr__(self): return f"Cell[{self.indexes}]" class HeaderCell(Cell): """ Column Header cell, a special cell that stores also a name and a size member. """ __slots__ = ["size", "name", "is_first", "sum_limit_num"] def __init__(self, name, limitNum=1): super(HeaderCell, self).__init__(limitNum) self.size = 0 self.name = name self.is_first = False self.sum_limit_num = 0 class DancingLinksMatrix: """ Dancing Links sparse matrix implementation. It stores a circular doubly linked list of 1s, and another list of column headers. Every cell points to its upper, lower, left and right neighbors in a circular fashion. """ def __init__(self, columns): """ Creates a DL_Matrix. :param columns: it can be an integer or an iterable. If columns is an integer, columns columns are added to the matrix, named C0,...,CN where N = columns -1. If columns is an iterable, the number of columns and the names are deduced from the iterable, else TypeError is raised. The iterable may yield the names, or a tuple (name,primary). primary is a bool value that is True if the column is a primary one. If not specified, is assumed that the column is a primary one. :raises TypeError, if columns is not a number neither an iterable. """ self.header = HeaderCell("<H>") self.header.is_first = True self.rows = self.cols = 0 self.col_list = [] self._create_column_headers(columns) def _create_column_headers(self, columns): if isinstance(columns, int): columns = int(columns) column_names = ((f"C{i}", 1) for i in range(columns)) else: try: column_names = iter(columns) except TypeError: raise TypeError("Argument is not valid") prev = self.header # links every column in a for loop for name in column_names: primary = True if isinstance(name, tuple) or isinstance(name, list): name, limitNum = name else: limitNum = 1 cell = HeaderCell(name, limitNum) cell.indexes = (-1, self.cols) cell.is_first = False self.col_list.append(cell) if primary: prev.R = cell cell.L = prev prev = cell self.cols += 1 prev.R = self.header self.header.L = prev def add_sparse_row(self, row, already_sorted=False): """ Adds a sparse row to the matrix. The row is in format [ind_0, ..., ind_n] where 0 <= ind_i < dl_matrix.ncols. If called after end_add is executed, CannotAddRowsError is raised. :param row: a sequence of integers indicating the 1s in the row. :param already_sorted: True if the row is already sorted, default is False. Use it for performance optimization. :raises CannotAddRowsError if end_add was already called. """ if self.col_list is None: raise CannotAddRowsError() prev = None start = None if not already_sorted: row = sorted(row) cell = None for ind in row: if isinstance(ind, int): ind = (ind, 1) cell = Cell(ind[1]) cell.indexes = (self.rows, ind[0]) if prev: prev.R = cell cell.L = prev else: start = cell col = self.col_list[ind[0]] # link the cell with the previous one and with the right column # cells. last = col.U last.D = cell cell.U = last col.U = cell cell.D = col cell.C = col col.size += 1 prev = cell col.sum_limit_num += ind[1] start.L = cell cell.R = start self.rows += 1 def end_add(self): """ Called when there are no more rows to be inserted. Not strictly necessary, but it can save some memory. """ self.col_list = None def min_column(self): """ Returns the column header of the column with the minimum number of 1s. :return: A column header. :raises: EmptyDLMatrix if the matrix is empty. """ # noinspection PyUnresolvedReferences if self.header.R.is_first: raise EmptyDLMatrix() col_min = self.header.R for col in iterate_cell(self.header, 'R'): if not col.is_first and col.size < col_min.size: col_min = col return col_min def random_column(self): """ Returns a random column header. (The matrix header is never returned) :return: A column header. :raises: EmptyDLMatrix if the matrix is empty. """ col = self.header.R if col is self.header: raise EmptyDLMatrix() n = random.randint(0, self.cols - 1) for _ in range(n): col = col.R if col.is_first: col = col.R return col def __str__(self): names = [] m = np.zeros((self.rows, self.cols), dtype=np.uint8) rows, cols = set(), [] for col in iterate_cell(self.header, 'R'): cols.append(col.indexes[1]) # noinspection PyUnresolvedReferences names.append(col.name) for cell in iterate_cell(col, 'D'): ind = cell.indexes rows.add(ind[0]) m[ind] = 1 m = m[list(rows)][:, cols] return " ".join([", ".join(names), str(m)]) @staticmethod def coverRow(r, isadd=False): for j in iterate_cell(r, 'R'): if j.C.limit_num < j.limit_num: return False for j in iterate_cell(r, 'R'): j.D.U = j.U j.U.D = j.D j.C.size -= 1 j.C.sum_limit_num -= j.limit_num if isadd: j.C.limit_num -= j.limit_num return True @staticmethod def checkCover(c): subLimitNum = {} for i in iterate_cell(c, 'D'): for j in iterate_cell(i, 'R'): idx = j.indexes[1] if idx not in subLimitNum: subLimitNum[idx] = 0 subLimitNum[idx] += j.limit_num if j.C.sum_limit_num - subLimitNum[idx] < j.C.limit_num: return False return True @staticmethod def cover(c, isadd=False): """ Covers the column c by removing the 1s in the column and also all the rows connected to them. :param c: The column header of the column that has to be covered. """ # print("Cover column", c.name) c.R.L = c.L c.L.R = c.R for i in iterate_cell(c, 'D'): DancingLinksMatrix.coverRow(i, isadd) return True @staticmethod def uncoverRow(r, isadd=False): for j in iterate_cell(r, 'L'): j.C.sum_limit_num += j.limit_num j.C.size += 1 j.D.U = j.U.D = j if isadd: j.C.limit_num += j.limit_num return True @staticmethod def uncover(c, isadd=False): """ Uncovers the column c by readding the 1s in the column and also all the rows connected to them. :param c: The column header of the column that has to be uncovered. """ # print("Uncover column", c.name) for i in iterate_cell(c, 'U'): DancingLinksMatrix.uncoverRow(i, isadd) c.R.L = c.L.R = c def iterate_cell(cell, direction): cur = getattr(cell, direction) while cur is not cell: yield cur cur = getattr(cur, direction) # TODO to be completed class MatrixDisplayer: def __init__(self, matrix): dic = {} for col in iterate_cell(matrix.header, 'R'): dic[col.indexes] = col for col in iterate_cell(matrix.header, 'R'): first = col.D dic[first.indexes] = first for cell in iterate_cell(first, 'D'): if cell is not col: dic[cell.indexes] = cell self.dic = dic self.rows = matrix.rows self.cols = matrix.cols def print_matrix(self): m = {} for i in range(-1, self.rows): for j in range(0, self.cols): cell = self.dic.get((i, j)) if cell: if i == -1: m[0, 2 * j] = cell.name+','+str(cell.limit_num) else: m[2 * (i + 1), 2 * j] = "X"+','+str(cell.limit_num) for i in range(-1, self.rows * 2): for j in range(0, self.cols * 2): print(m.get((i, j), " "), end="") print() if __name__ == "__main__": def from_dense(row): return [i for i, el in enumerate(row) if el] r = [from_dense([1, 0, 0, 1, 0, 0, 1]), from_dense([1, 0, 0, 1, 0, 0, 0]), from_dense([0, 0, 0, 1, 1, 0, 1]), from_dense([0, 0, 1, 0, 1, 1, 0]), from_dense([0, 1, 1, 0, 0, 1, 1]), from_dense([0, 1, 0, 0, 0, 0, 1])] d = DancingLinksMatrix("1234567") for row in r: d.add_sparse_row(row, already_sorted=True) d.end_add() p = MatrixDisplayer(d) p.print_matrix() # print(d.rows) # print(d.cols) # print(d) mc = d.min_column() # print(mc) d.cover(mc) # print(d) p.print_matrix()
还有这个：
""" Implementation of Donald Knuth's Algorithm X (http://arxiv.org/abs/cs/0011047). """ from dlmatrix import DancingLinksMatrix, iterate_cell, MatrixDisplayer import string __author__ = 'FunCfans' testRow = [0 ,62 ,63 ,25 ,37 ,52 ,32 ,54 ,40 ,17 ,34 ,19 ,24 ,13] class AlgorithmX: """Callable object implementing the Algorithm X.""" def __init__(self, matrix, callback, choose_min=True): """ Creates an Algorithm_X object that solves the problem encoded in matrix. :param matrix: The DL_Matrix instance. :param callback: The callback called on every solution. callback has to be a function receiving a dict argument {row_index: linked list of the row}, and can return a bool value. The solver keeps going on until the callback returns a True value. :param choose_min: If True, the column with the minimum number of 1s is chosen at each iteration, if False a random column is chosen. """ self.sol_dict = {} self.stop = False self.matrix = matrix self.callback = callback self.choose_min = choose_min self.deduce_cnt = 0 self.depth = 0 self.last_matrix = None self.delta_file = None def __call__(self): """Starts the search.""" #self.delta_file = open('delta_matrix.txt','w',encoding='utf-8') #self._print(self.matrix.header, 'start') self._search(0) #self.delta_file.close() def _print(self, currrow, op): self.deduce_cnt += 1 f = open('step/' + 'step%04d_' % (self.deduce_cnt) + op + '_' + str(currrow.indexes) + '_depth%d.txt'%(self.depth),'w',encoding='utf-8') printrow = {} rowcontent = '' content = 'curr ' + op + ' row : ' + str(currrow.indexes) + ' ' content = '' for col in iterate_cell(self.matrix.header, 'R'): content += 'col name : '+col.name+' col limit : ' + str(col.limit_num) for row in iterate_cell(col, 'D'): content += ' ' + str((row.indexes[0],row.limit_num)) + ',' printrow[row.indexes[0]] = row content += ' ' content += ' ' for k,v in printrow.items(): content += 'row %d : '%(k) + str((v.limit_num, v.indexes[1])) + '-->' qv = v.R while(qv != v): content += str((qv.limit_num, qv.indexes[1])) + '-->' qv = qv.R content += str((qv.limit_num, qv.indexes[1])) + ' ' f.write(content) f.close() if self.last_matrix == None: self.last_matrix = {} for col in iterate_cell(self.matrix.header, 'R'): self.last_matrix[(col.name, col.indexes[0])] = collist = set() for row in iterate_cell(col, 'D'): collist.add(row.indexes[0]) else: curr_matrix = {} for col in iterate_cell(self.matrix.header, 'R'): curr_matrix[(col.name, col.indexes[0])] = collist = set() for row in iterate_cell(col, 'D'): collist.add(row.indexes[0]) add_col = set() addv = set(curr_matrix.keys()).difference(set(self.last_matrix.keys())) if len(addv) > 0: self.delta_file.write('add_col : '+str(addv) + ' ') delv = set(self.last_matrix.keys()).difference(set(curr_matrix.keys())) if len(delv) > 0: self.delta_file.write('delete_col : '+str(delv) + ' ') for k,v in curr_matrix.items(): if k not in self.last_matrix:continue addv = v.difference(self.last_matrix[k]) if len(addv) > 0: self.delta_file.write('add_col_'+str(k)+' : '+str(addv)) self.delta_file.write(' ') for k,v in self.last_matrix.items(): if k not in curr_matrix:continue delv = self.last_matrix[k].difference(v) if len(delv) > 0: self.delta_file.write('delete_col_'+str(k)+' : '+str(delv)) self.delta_file.write(' ') self.delta_file.write(' ') self.last_matrix = curr_matrix def _search(self, k): # print(f"Size: {k}") # k is depth # print(f"Solution: {self.sol_dict}") # print("Matrix:") # print(self.matrix) if self.matrix.header.R == self.matrix.header: # matrix is empty, solution found if self.callback(self._create_sol(k)): self.stop = True return if self.choose_min: col = self.matrix.min_column() else: col = self.matrix.random_column() # cover column col # row = col.D rows = [] for row in iterate_cell(col, 'D'): rows.append(row) rows.sort(key=lambda x:x.limit_num,reverse=True) self.depth += 1 for row in rows: if col.limit_num < row.limit_num:continue if col.sum_limit_num < col.limit_num:break isValid = True for j in iterate_cell(row, 'R'): if j.C.limit_num < j.limit_num: isValid = False break if not isValid: continue self.sol_dict[k] = row col.sum_limit_num -= row.limit_num col.limit_num -= row.limit_num row.D.U = row.U row.U.D = row.D col.size -= 1 if col.limit_num == 0: self.matrix.cover(col) for j in iterate_cell(row, 'R'): j.C.sum_limit_num -= j.limit_num j.C.limit_num -= j.limit_num j.D.U = j.U j.U.D = j.D j.C.size -= 1 if j.C.limit_num == 0: self.matrix.cover(j.C) execYou = True for j in iterate_cell(self.matrix.header, 'R'): if j.limit_num > j.sum_limit_num: execYou = False break if execYou: self._search(k + 1) if self.stop: return for j in iterate_cell(row, 'L'): if j.C.limit_num == 0: self.matrix.uncover(j.C) j.C.limit_num += j.limit_num if col.limit_num == 0: self.matrix.uncover(col) col.limit_num += row.limit_num del self.sol_dict[k] # uncover columns for row in rows[::-1]: for j in iterate_cell(row, 'L'): j.C.size += 1 j.D.U = j.U.D = j j.C.sum_limit_num += j.limit_num col.size += 1 row.D.U = row.U.D = row col.sum_limit_num += row.limit_num self.depth -= 1 # def _create_sol(self, k): # creates a solution from the inner dict sol = {} for key, row in self.sol_dict.items(): if key >= k: continue tmp_list = [row.C.name] tmp_list.extend(r.C.name for r in iterate_cell(row, 'R')) sol[row.indexes[0]] = tmp_list return sol # def main(): from_dense = (lambda row:[i for i, el in enumerate(row) if el]) rows = [from_dense([0, 0, 1, 0, 1, 1, 0]), from_dense([1, 0, 0, 1, 0, 0, 1]), from_dense([0, 1, 1, 0, 0, 1, 0]), from_dense([1, 0, 0, 1, 0, 0, 0]), from_dense([0, 1, 0, 0, 0, 0, 1]), from_dense([0, 0, 0, 1, 1, 0, 1])] size = max(max(rows, key=max)) + 1 d = DancingLinksMatrix(string.ascii_uppercase[:size]) for row in rows: d.add_sparse_row(row, already_sorted=True) AlgorithmX(d, print)() # if __name__ == "__main__": main()