fork(2) download
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. import Queue
  4. import argparse
  5. import os
  6. import random
  7. import subprocess
  8. import sys
  9. import threading
  10. import time
  11.  
  12. class Tester(threading.Thread):
  13. def __init__(self, testee, answer_queue, counter_queue):
  14. threading.Thread.__init__(self)
  15. self.testee = testee
  16. self.answer_queue = answer_queue
  17. self.counter_queue = counter_queue
  18. def run(self):
  19. while True:
  20. ans = self.answer_queue.get()
  21. counter = 1
  22. p = subprocess.Popen([sys.executable, self.testee],
  23. stdin=subprocess.PIPE, stdout=subprocess.PIPE)
  24. while True:
  25. guess = p.stdout.readline().strip()
  26. bulls = sum(1 for i, j in zip(ans, guess) if i == j)
  27. cows = sum(1 for i, j in zip(ans, guess) if i != j and j in ans)
  28. score = str(bulls) + 'a' + str(cows) + 'b' + '\n'
  29. p.stdin.write(score)
  30. if bulls == 4:
  31. self.answer_queue.task_done()
  32. self.counter_queue.put(counter)
  33. counter = 1
  34. break
  35. counter += 1
  36. if self.answer_queue.empty():
  37. break
  38.  
  39. def get_args():
  40. args_parser = argparse.ArgumentParser(
  41. description='Test bulls-and-cows guesser\'s performance.')
  42. args_parser.add_argument('testee' , metavar='TESTEE',
  43. help='the guesser you want to evaluate')
  44. args_parser.add_argument('-j', '--jobs', metavar='N', type=int, default=1,
  45. help='allow N jobs at once')
  46. return args_parser.parse_args()
  47.  
  48. def get_all_answers(shuffle):
  49. answer_candidates = [str(i) + str(j) + str(k) + str(l) \
  50. for i in xrange(0, 10) for j in xrange(0, 10) \
  51. for k in xrange(0, 10) for l in xrange(0, 10) \
  52. if len(set(str(i) + str(j) + str(k) + str(l))) == 4]
  53.  
  54. if shuffle:
  55. random.shuffle(answer_candidates)
  56.  
  57. return answer_candidates
  58.  
  59. def main():
  60. args = get_args()
  61. answer_candidates = get_all_answers(True)
  62.  
  63. answer_queue = Queue.Queue()
  64. counter_queue = Queue.Queue()
  65. for ans in answer_candidates:
  66. answer_queue.put(ans)
  67. for i in xrange(args.jobs):
  68. t = Tester(args.testee, answer_queue, counter_queue)
  69. t.daemon = True
  70. t.start()
  71.  
  72. answer_queue.join()
  73. while True:
  74. if counter_queue.qsize() == len(answer_candidates):
  75. break
  76.  
  77. counter = []
  78. for i in xrange(len(answer_candidates)):
  79. counter.append(counter_queue.get())
  80.  
  81. print 'max: ', max(counter)
  82. print 'min: ', min(counter)
  83. print 'avg: ', float(sum(counter)) / len(counter)
  84.  
  85. if __name__ == '__main__':
  86. main()
Not running #stdin #stdout 0s 0KB
stdin
Standard input is empty
stdout
Standard output is empty