Python: implementacja algorytmu Astar

Sep 02 2020

Zaimplementowałem algorytm Astar dla problemu z sędzią online, odnoszącego się do labiryntu z podanymi pozycjami początkowymi i końcowymi wraz z siatką reprezentującą labirynt. Podaję długość ścieżki wraz z samą ścieżką. Poniżej przedstawiono implementację w Pythonie przy użyciu odległości euklidesowej:

import heapq, math, sys

infinity = float('inf')

class AStar():

    def __init__(self, start, grid, height, width):
        self.start, self.grid, self.height, self.width = start, grid, height, width

    class Node():
        def __init__(self, position, fscore=infinity, gscore=infinity, parent = None):
            self.fscore, self.gscore, self.position, self.parent = fscore, gscore, position, parent
            
        def __lt__(self, comparator):
            return self.fscore < comparator.fscore

    def heuristic(self, end, distance = "Euclidean"):
        (x1, y1), (x2, y2) = self.start, end
        if (distance == "Manhattan"):
            return abs(x1 - x2) + abs(y1 - y2)
        return math.sqrt((x2 - x1)**2 + (y2 - y1)**2)

    def nodeNeighbours(self, pos):
        (x, y) = pos
        return [(dx, dy) for (dx, dy) in [(x + 1, y), (x - 1, y), (x, y + 1), (x, y - 1)] if 0 <= dx < self.width and 0 <= dy < self.height and self.grid[dy][dx] == 0]

    def getPath(self, endPoint):
        current, path = endPoint, []
        while current.position != self.start:
            path.append(current.position)
            current = current.parent
        path.append(self.start)
        return list(reversed(path))

    def computePath(self, end):
        openList, closedList, nodeDict = [], [], {}
        currentNode = AStar.Node(self.start, fscore=self.heuristic(end), gscore = 0)
        heapq.heappush(openList, currentNode)
        while openList:
            currentNode = heapq.heappop(openList)
            if currentNode.position == end:
                return self.getPath(currentNode)
            else:
                closedList.append(currentNode)
                neighbours = []
                for toCheck in self.nodeNeighbours(currentNode.position):
                    if toCheck not in nodeDict.keys():
                        nodeDict[toCheck] = AStar.Node(toCheck)
                        neighbours.append(nodeDict[toCheck])
                
                for neighbour in neighbours:
                    newGscore = currentNode.gscore + 1
                    if neighbour in openList and newGscore < neighbour.gscore:
                        openList.remove(neighbour)
                    if newGscore < neighbour.gscore and neighbour in closedList:
                        closedList.remove(neighbour)
                    if neighbour not in openList and neighbour not in closedList:
                        neighbour.gscore = newGscore
                        neighbour.fscore = neighbour.gscore + self.heuristic(neighbour.position)
                        neighbour.parent = currentNode
                        heapq.heappush(openList, neighbour)
                    heapq.heapify(openList)
        return None
        
if __name__ == '__main__':
    
    sys.stdin = open('input.txt', 'r')
    sys.stdout = open('output.txt', 'w')
    
    matrix = [[int(num) for num in line.split()] for line in sys.stdin]
    size = matrix.pop(0)
    coordinates = matrix.pop(0)
    n, m = size[0], size[1]
    x1, y1, y2, x2 = coordinates[0], coordinates[1], coordinates[2], coordinates[3]
    path = AStar((x1-1, y1-1), matrix, n, m).computePath((y2-1, x2-1))
    print(len(path))
    for pos in path:
        print(pos[0] + 1, pos[1] + 1)

Odpowiedzi

5 Carcigenicate Sep 02 2020 at 21:24
self.start, self.grid, self.height, self.width = start, grid, height, width

Nie umieściłbym ich wszystkich w tej samej linii. Myślę, że znacznie łatwiej byłoby czytać w wielu wierszach:

self.start = start
self.grid = grid
self.height = height
self.width = width

Prawdopodobnie miałbym Nodeklasę jako najwyższy poziom zamiast zagnieżdżonej. Nie sądzę, żebyś wiele zyskał mając to w środku AStar. Możesz nazwać go _Nodetak, aby był „prywatny dla modułu”, tak aby próba zaimportowania go do innego pliku potencjalnie wywołała ostrzeżenia.

W Node„s __lt__realizacji, nie nazwałbym drugi parametr comparator. Porównywacz to coś, co porównuje, podczas gdy w tym przypadku jest to po prostu kolejny węzeł. other_nodelub coś byłoby bardziej odpowiednie.


W heuristic, osobiście skorzystałbym z elsetam:

if (distance == "Manhattan"):
    return abs((x1 - x2) + abs(y1 - y2))
else:
    return math.sqrt((x2 - x1)**2 + (y2 - y1)**2)

Dzięki temu jest jaśniejsze, że zostanie wykonana tylko jedna z linii. Osobiście pomijam tylko elsew takim przypadku, jeśli ifbyło to sprawdzanie warunków wstępnych „wczesnego wyjścia”, i chcę uniknąć zagnieżdżania całej reszty funkcji w bloku. Nie stanowi to jednak problemu.


nodeNeighbors( co powinno byćnode_neighbors ) byłoby czystsze i podzielone na kilka linii:

def nodeNeighbours(self, pos):
    (x, y) = pos
    return [(dx, dy)
            for (dx, dy) in [(x + 1, y), (x - 1, y), (x, y + 1), (x, y - 1)]
            if 0 <= dx < self.width and 0 <= dy < self.height and self.grid[dy][dx] == 0]

Myślę, że dzięki temu o wiele łatwiej jest zobaczyć, co się w nim dzieje.


W wielu miejscach przypisujesz dwie lub więcej zmiennych w jednym wierszu:

(x1, y1), (x2, y2) = self.start, end
current, path = endPoint, []
openList, closedList, nodeDict = [], [], {}
x1, y1, y2, x2 = coordinates[0], coordinates[1], coordinates[2], coordinates[3]

Rozbiję to. Zwłaszcza, gdy dojdziesz do 3+ w wierszu, aby czytelnik mógł zobaczyć, która zmienna pasuje do jakiej wartości, będzie musiał liczyć od lewej strony, zamiast po prostu sprawdzać, co jest po obu stronach a =.


W computePathwydaje się, że closedListpowinien być zestaw. Nie wygląda na to, że porządek ma z nim znaczenie i neighbour in closedListbędzie szybszy z zestawem niż z listą. Wygląda jednak na openListto, że musi to być lista, ponieważ jest przekazywana do heapify.


Nie sądzę, żebym ponownie przypisał stdini stdout. Ponowne przypisanie stdinwydaje się całkowicie niepotrzebne, a zmiana stdoututrudni późniejsze debugowanie przy użyciu printinstrukcji. Niekoniecznie chcesz, aby cały drukowany tekst był wysyłany do pliku.

W razie potrzeby możesz określić, do jakiego pliku chcesz drukować podczas drukowania:

with open('output.txt', 'w') as out_f:
    print("To file!", file=out_f)