Source Code for Module networkx.queues

  1  """ 
  2  Helper queues for use in graph searching. 
  3   
  4   - LIFO:          Last in first out queue (stack) 
  5   - FIFO:          First in first out queue 
  6   - Priority(fcn): Priority queue with items are sorted by fcn 
  7   - Random:        Random queue 
  8   
  9   - q.append(item)  -- add an item to the queue 
 10   - q.extend(items) -- equivalent to: for item in items: q.append(item) 
 11   - q.pop()         -- return the top item from the queue 
 12   - len(q)          -- number of items in q (also q.__len()) 
 13  """ 
 14  __author__ = """Aric Hagberg (hagberg@lanl.gov)""" 
 15  __date__ = "$Date: 2005-03-30 16:56:28 -0700 (Wed, 30 Mar 2005) $" 
 16  __credits__ = """""" 
 17  __revision__ = "$Revision: 911 $" 
 18  #    Copyright (C) 2004,2005 by  
 19  #    Aric Hagberg <hagberg@lanl.gov> 
 20  #    Dan Schult <dschult@colgate.edu> 
 21  #    Pieter Swart <swart@lanl.gov> 
 22  #    Distributed under the terms of the GNU Lesser General Public License 
 23  #    http://www.gnu.org/copyleft/lesser.html 
 24  import bisect 
 25  import random 
 26   
 27 -class LIFO(list): 
 28      """ 
 29      """ 
 30 -    def __init__(self): 
 31          list.__init__(self) 
 32   
 33 -class FIFO(list): 
 34      """ 
 35      """ 
 36 -    def __init__(self): 
 37          list.__init__(self) 
 38 -    def pop(self): 
 39          item = list.pop(self,0) 
 40          return item 
 41   
 42 -class Random(list): 
 43      """ 
 44      """ 
 45 -    def __init__(self): 
 46          list.__init__(self) 
 47 -    def pop(self): 
 48          index=random.randint(0,len(self)-1) 
 49          item = list.pop(self,index) 
 50          return item 
 51   
 52   
 53 -class Priority: 
 54      """ 
 55      """ 
 56 -    def __init__(self, f=lambda x: x): 
 57          self.L=[] 
 58          self.f=f 
 59 -    def append(self, item): 
 60          bisect.insort(self.L, (self.f(item), item)) 
 61 -    def __len__(self): 
 62          return len(self.L) 
 63 -    def extend(self, items): 
 64          for item in items: self.append(item) 
 65 -    def pop(self): 
 66          return self.L.pop()[1] 
 67 -    def smallest(self): 
 68          return self.L.pop(0)[1] 
 69   
 70   
 71  # search queue types 
 72  # for the generic search class we need an "update" function so let's 
 73  # make a new class that adds an update method to the Queues classes 
 74   
 75 -class DFS(LIFO): 
 76      """ 
 77      Depth first search queue 
 78      """ 
 79 -    def __init__(self): 
 80          LIFO.__init__(self) 
 81   
 82 -    def update(self, item): 
 83          # DFS update rule for queue (self is the queue) 
 84          # Find the edge on the fringe with same target and delete 
 85          (source, target)=item  
 86          olditem=[(u,v) for (u,v) in self if v==target][0] 
 87          list.remove(self,olditem) 
 88          list.append(self,item) 
 89   
 90   
 91 -class BFS(FIFO): 
 92      """ 
 93      Breadth first search queue 
 94      """ 
 95 -    def __init__(self): 
 96          FIFO.__init__(self) 
 97   
 98 -    def update(self, item): 
 99          # BFS update rule for queue, do nothing 
100          pass 
101   
102   
103 -class RFS(Random): 
104      """ 
105      Random search queue 
106      """ 
107 -    def __init__(self): 
108          Random.__init__(self) 
109   
110 -    def update(self, item): 
111          # RFS update rule for queue, do nothing 
112          pass 
113   
114 -def _test_suite(): 
115      import doctest 
116      suite = doctest.DocFileSuite('tests/queues.txt',package='networkx') 
117      return suite 
118   
119   
120  if __name__ == "__main__": 
121      import os 
122      import sys 
123      import unittest 
124      if sys.version_info[:2] < (2, 4): 
125          print "Python version 2.4 or later required for tests (%d.%d detected)." %  sys.version_info[:2] 
126          sys.exit(-1) 
127      # directory of networkx package (relative to this) 
128      nxbase=sys.path[0]+os.sep+os.pardir 
129      sys.path.insert(0,nxbase) # prepend to search path 
130      unittest.TextTestRunner().run(_test_suite()) 
131