pacai.util.containers

  1import abc
  2import heapq
  3import typing
  4
  5T = typing.TypeVar('T')
  6""" A generic type to ensure type consistency for all the container classes. """
  7
  8class FringeContainer(abc.ABC, typing.Generic[T]):
  9    """
 10    A generic container base class that is useful for a search fringe
 11    (but can be used for any sort of storage).
 12    """
 13
 14    def __init__(self) -> None:
 15        self._items: list[T] = []
 16        """
 17        The underlying storage for our container's items.
 18        We can just use a normal list as long as we are careful about how we work with it.
 19        """
 20
 21    def is_empty(self) -> bool:
 22        """ Returns True if the container is empty. """
 23
 24        return (len(self._items) == 0)
 25
 26    def __len__(self) -> int:
 27        """ Override the len() operator to get the size of the container. """
 28
 29        return len(self._items)
 30
 31    @abc.abstractmethod
 32    def push(self, item: T) -> None:
 33        """ Add an item to this container. """
 34
 35    @abc.abstractmethod
 36    def pop(self) -> T:
 37        """ Remove the next item from this container. """
 38
 39class Stack(FringeContainer[T]):
 40    """
 41    A container with a last-in-first-out (LIFO) queuing policy.
 42    See https://en.wikipedia.org/wiki/Stack_(abstract_data_type) .
 43    """
 44
 45    def push(self, item: T) -> None:
 46        """ Push an item onto the stack. """
 47
 48        self._items.append(item)
 49
 50    def pop(self) -> T:
 51        """ Pop the most recently pushed item from the stack. """
 52
 53        return self._items.pop()
 54
 55class Queue(FringeContainer[T]):
 56    """
 57    A container with a first-in-first-out (FIFO) queuing policy.
 58    See: https://en.wikipedia.org/wiki/Queue_(abstract_data_type) .
 59    """
 60
 61    def push(self, item: T) -> None:
 62        """ Enqueue the item into the queue. """
 63
 64        self._items.insert(0, item)
 65
 66    def pop(self) -> T:
 67        """ Dequeue the earliest enqueued item still in the queue. """
 68
 69        return self._items.pop()
 70
 71class PriorityQueue(FringeContainer[T]):
 72    """
 73    A container with a queuing policy that prioritizes objects with lower priority..
 74    See: https://en.wikipedia.org/wiki/Priority_queue .
 75
 76    Each inserted item has a priority associated with it,
 77    and the user is usually interested in quick retrieval of the lowest-priority item in the queue.
 78    This data structure allows O(1) access to the lowest-priority item.
 79
 80    Note that this PriorityQueue does not allow you to change the priority of an item.
 81    However, you may insert the same item multiple times with different priorities.
 82    """
 83
 84    def push(self, item: T, priority: float) -> None:  # type: ignore  # pylint: disable=arguments-differ
 85        """ Enqueue the item into the priority queue. """
 86
 87        pair = (priority, item)
 88        heapq.heappush(self._items, pair)  # type: ignore
 89
 90    def pop(self) -> T:
 91        """ Dequeue the earliest enqueued item still in the priority queue. """
 92
 93        (_, item) = heapq.heappop(self._items)  # type: ignore
 94        return item  # type: ignore
 95
 96@typing.runtime_checkable
 97class PriorityFunction(typing.Protocol):
 98    """
 99    A function that assigns a priority value to an object being inserted into the priority queue.
100    """
101
102    def __call__(self, item: typing.Any) -> float:
103        ...
104
105class PriorityQueueWithFunction(PriorityQueue[T]):
106    """
107    Implements a priority queue with the same push/pop signature of the Queue and the Stack classes.
108    This is designed for drop-in replacement for those two classes.
109    The caller has to provide a priority function, which extracts each item's priority.
110    """
111
112    def __init__(self, priority_func: PriorityFunction) -> None:
113        super().__init__()
114
115        self._priority_func: PriorityFunction = priority_func
116        """ The function to get a priority for each item in this container. """
117
118    def push(self, item: T) -> None:  # type: ignore  # pylint: disable=arguments-differ
119        """
120        Adds an item to the queue with priority from the priority function
121        """
122
123        super().push(item, self._priority_func(item))