traviskaufman/cats_vs_dogs.py

## cats_vs_dogs.py
"""
I apologize this isn't more well-documented :/
"""

#!/usr/bin/env python

"""
Plays the game of cat vs. dog
"""
from sys import stdin
from multiprocessing import cpu_count
import threading
import re


class Viewer(object):
    """Simple class for an audience member."""
    who_stays = None
    who_leaves = None

    def __init__(self, who_stays, who_leaves):
        self.who_stays = who_stays
        self.who_leaves = who_leaves
        self._utility = 0

    def will_continue_to_watch(self):
        """Determines based on the audience member's utility whether s/he will
        continue to watch the reality TV show.
        """
        return self._utility > 0

    def observe_outcome(self, remaining_animals):
        """Looks at the outcome of the results and Calculates his/her utility
        based on it. It will basically only set utility positive when
        """
        if (self.who_stays in remaining_animals) and \
           (self.who_leaves not in remaining_animals):
            self._utility = self._utility + 1


def get_successors(current_animals):
    """Returns a list of lists containing all of the different possible
    outcomes that would result from removing one animal. Each outcome would
    have a different animal removed.
    """
    lnth = len(current_animals)
    lr = range(lnth)
    successors = [None for _ in lr]
    for i in lr:
        successors[i] = [current_animals[idx] for idx in lr if idx != i]

    return successors


def create_initial_animals(num_dogs, num_cats):
    """Creates the initial list of cats and dogs"""
    dogs = [("D" + str(i)) for i in range(1, num_dogs + 1)]
    cats = [("C" + str(i)) for i in range(1, num_cats + 1)]
    dogs.extend(cats)
    return dogs


def get_num_viewers_remaining(remaining_animals, viewers):
    """From a list of audience members, figure out based on a given state of
    remaining animals the number of audience members that would continue
    watching the show.
    """
    num_remaining = 0
    for viewer in viewers:
        viewer.observe_outcome(remaining_animals)
        if viewer.will_continue_to_watch():
            num_remaining = num_remaining + 1

    return num_remaining


def get_max_viewers(animals, viewers):
    """Determines the max amount of viewers that could be retained based on
    which animals are kept on and/or voted off. Basically a breadth-wise
    minimax with an alpha-beta pruning-like succession scheme.
    """
    # Edge Case: 1 Viewer, 1 Cat, 1 Dog.
    if viewers == 1 and len(animals) <= 2:
        return 0

    new_list = [[i] for i in animals]
    best = None

    # This could be done recursively but doing it dynamically saves stack space
    max_viewers = 0
    while (len(new_list[0]) < len(animals)):
        for state, idx in zip(new_list, range(len(new_list))):
            viewers_remaining = get_num_viewers_remaining(animals, viewers)
            if viewers_remaining > max_viewers:
                max_viewers, best = viewers_remaining, new_list[idx]
            else:  # we found the max
                break

        new_list = [best + [i] for i in animals if i not in best]

    return max_viewers


def simulate_games(stdinput=None):
    """Faux simulation of the game with test data."""
    game_components = stdinput.split("\n")
    # If remove a blank '' entry if present
    if game_components[len(game_components) - 1] == '':
        game_components = game_components[:-1]

    num_tests = int(game_components.pop(0))
    ranges = partition_offsets(game_components)
    results = []

    # UGH Python Threading -_-
    max_threads = max(cpu_count() + 1, 3)
    thread_pool = []
    results_mutex = threading.Semaphore()
    while num_tests > 0:
        for i in range(min(max_threads, num_tests)):
            start, lnth = ranges[i]
            partition = game_components[start:lnth]

            thread = threading.Thread(
                target=run_game, name=str(i),
                args=(partition, results, results_mutex)
            )
            thread_pool.append(thread)
            thread.start()

            for thread in thread_pool:
                thread.join()

        num_tests = num_tests - max_threads

    return [res for tid, res in sorted(results)]


def run_game(game_components, results_list, results_mutex):
    """Run inside a thread."""
    max_viewers = 0
    # Edge Case: 1 Viewer, 1 Cat, 1 Dog. No one will be happy.
    if game_components[0] != "1 1 1":
        num_cats, num_dogs, num_viewers = tuple(
            [int(i) for i in game_components.pop(0).split(" ") if i != '']
        )

        initial_animals = create_initial_animals(num_dogs, num_cats)
        viewers = []
        for i in range(num_viewers):
            stay, leave = tuple(game_components.pop(0).split(" "))
            viewers.append(Viewer(stay, leave))

        max_viewers = get_max_viewers(initial_animals, viewers)

    results_mutex.acquire()
    # Hacky: Return a tuple with the thread name (an int ID) so we can sort
    # results
    results_list.append((int(threading.currentThread().name), max_viewers))
    results_mutex.release()


def partition_offsets(split_txt):
    """Partitions list into different game test cases that can be run in
    parallel. Returns a list of tuples showing the ranges that need to be
    extracted from the list of game_components so it can be run in parallel.
    """
    test_data = [n for n in split_txt if re.match(r'\d+ \d+ \d+', n)]
    start = 0
    ranges = []
    for datum in test_data:
        lnth = int(datum.split(" ").pop()) + start + 1
        ranges.append((start, lnth))
        start = lnth

    return ranges


def main():
    """MAGNETS!!"""
    stdinput = stdin.read()

    results = simulate_games(stdinput)
    if results:
        for result in results:
            print result


if __name__ == "__main__":
    main()

## description.md

      
    Raw
  

              description.md
            
          
    Usage

Test Accuracy and Timing

$ time python2.6 cats_vs_dogs.py < file_name.txt
From Spotify

The latest reality show has hit the TV: “Cat vs. Dog”. In this show, a bunch of cats and dogs compete for the very prestigious Best Pet Ever title. In each episode, the cats and dogs get to show themselves off, after which the viewers vote on which pets should stay and which should be forced to leave the show.
Each viewer gets to cast a vote on two things: one pet which should be kept on the show, and one pet which should be thrown out. Also, based on the universal fact that everyone is either a cat lover (i.e. a dog hater) or a dog lover (i.e. a cat hater), it has been decided that each vote must name exactly one cat and exactly one dog.
Ingenious as they are, the producers have decided to use an advancement procedure which guarantees that as many viewers as possible will continue watching the show: the pets that get to stay will be chosen so as to maximize the number of viewers who get both their opinions satisfied. Write a program to calculate this maximum number of viewers.
Input

On the first line one positive number: the number of testcases, at most 100. After that per testcase:
One line with three integers c, d, v (1 ≤ c, d ≤ 100 and 0 ≤ v ≤ 500): the number of cats, dogs, and voters.
v lines with two pet identifiers each. The first is the pet that this voter wants to keep, the second is the pet that this voter wants to throw out. A pet identifier starts with one of the characters ‘C’ or ‘D’, indicating whether the pet is a cat or dog, respectively. The remaining part of the identifier is an integer giving the number of the pet (between 1 and c for cats, and between 1 and d for dogs). So for instance, “D42” indicates dog number 42.
Output

Per testcase:
One line with the maximum possible number of satisfied voters for the show.
Sample Input 1
2
1 1 2
C1 D1
D1 C1
1 2 4
C1 D1
C1 D1
C1 D2
D2 C1

Sample Output
1
3
	"""
	I apologize this isn't more well-documented :/
	"""

	#!/usr/bin/env python

	"""
	Plays the game of cat vs. dog
	"""
	from sys import stdin
	from multiprocessing import cpu_count
	import threading
	import re


	class Viewer(object):
	"""Simple class for an audience member."""
	who_stays = None
	who_leaves = None

	def __init__(self, who_stays, who_leaves):
	self.who_stays = who_stays
	self.who_leaves = who_leaves
	self._utility = 0

	def will_continue_to_watch(self):
	"""Determines based on the audience member's utility whether s/he will
	continue to watch the reality TV show.
	"""
	return self._utility > 0

	def observe_outcome(self, remaining_animals):
	"""Looks at the outcome of the results and Calculates his/her utility
	based on it. It will basically only set utility positive when
	"""
	if (self.who_stays in remaining_animals) and \
	(self.who_leaves not in remaining_animals):
	self._utility = self._utility + 1


	def get_successors(current_animals):
	"""Returns a list of lists containing all of the different possible
	outcomes that would result from removing one animal. Each outcome would
	have a different animal removed.
	"""
	lnth = len(current_animals)
	lr = range(lnth)
	successors = [None for _ in lr]
	for i in lr:
	successors[i] = [current_animals[idx] for idx in lr if idx != i]

	return successors


	def create_initial_animals(num_dogs, num_cats):
	"""Creates the initial list of cats and dogs"""
	dogs = [("D" + str(i)) for i in range(1, num_dogs + 1)]
	cats = [("C" + str(i)) for i in range(1, num_cats + 1)]
	dogs.extend(cats)
	return dogs


	def get_num_viewers_remaining(remaining_animals, viewers):
	"""From a list of audience members, figure out based on a given state of
	remaining animals the number of audience members that would continue
	watching the show.
	"""
	num_remaining = 0
	for viewer in viewers:
	viewer.observe_outcome(remaining_animals)
	if viewer.will_continue_to_watch():
	num_remaining = num_remaining + 1

	return num_remaining


	def get_max_viewers(animals, viewers):
	"""Determines the max amount of viewers that could be retained based on
	which animals are kept on and/or voted off. Basically a breadth-wise
	minimax with an alpha-beta pruning-like succession scheme.
	"""
	# Edge Case: 1 Viewer, 1 Cat, 1 Dog.
	if viewers == 1 and len(animals) <= 2:
	return 0

	new_list = [[i] for i in animals]
	best = None

	# This could be done recursively but doing it dynamically saves stack space
	max_viewers = 0
	while (len(new_list[0]) < len(animals)):
	for state, idx in zip(new_list, range(len(new_list))):
	viewers_remaining = get_num_viewers_remaining(animals, viewers)
	if viewers_remaining > max_viewers:
	max_viewers, best = viewers_remaining, new_list[idx]
	else: # we found the max
	break

	new_list = [best + [i] for i in animals if i not in best]

	return max_viewers


	def simulate_games(stdinput=None):
	"""Faux simulation of the game with test data."""
	game_components = stdinput.split("\n")
	# If remove a blank '' entry if present
	if game_components[len(game_components) - 1] == '':
	game_components = game_components[:-1]

	num_tests = int(game_components.pop(0))
	ranges = partition_offsets(game_components)
	results = []

	# UGH Python Threading -_-
	max_threads = max(cpu_count() + 1, 3)
	thread_pool = []
	results_mutex = threading.Semaphore()
	while num_tests > 0:
	for i in range(min(max_threads, num_tests)):
	start, lnth = ranges[i]
	partition = game_components[start:lnth]

	thread = threading.Thread(
	target=run_game, name=str(i),
	args=(partition, results, results_mutex)
	)
	thread_pool.append(thread)
	thread.start()

	for thread in thread_pool:
	thread.join()

	num_tests = num_tests - max_threads

	return [res for tid, res in sorted(results)]


	def run_game(game_components, results_list, results_mutex):
	"""Run inside a thread."""
	max_viewers = 0
	# Edge Case: 1 Viewer, 1 Cat, 1 Dog. No one will be happy.
	if game_components[0] != "1 1 1":
	num_cats, num_dogs, num_viewers = tuple(
	[int(i) for i in game_components.pop(0).split(" ") if i != '']
	)

	initial_animals = create_initial_animals(num_dogs, num_cats)
	viewers = []
	for i in range(num_viewers):
	stay, leave = tuple(game_components.pop(0).split(" "))
	viewers.append(Viewer(stay, leave))

	max_viewers = get_max_viewers(initial_animals, viewers)

	results_mutex.acquire()
	# Hacky: Return a tuple with the thread name (an int ID) so we can sort
	# results
	results_list.append((int(threading.currentThread().name), max_viewers))
	results_mutex.release()


	def partition_offsets(split_txt):
	"""Partitions list into different game test cases that can be run in
	parallel. Returns a list of tuples showing the ranges that need to be
	extracted from the list of game_components so it can be run in parallel.
	"""
	test_data = [n for n in split_txt if re.match(r'\d+ \d+ \d+', n)]
	start = 0
	ranges = []
	for datum in test_data:
	lnth = int(datum.split(" ").pop()) + start + 1
	ranges.append((start, lnth))
	start = lnth

	return ranges


	def main():
	"""MAGNETS!!"""
	stdinput = stdin.read()

	results = simulate_games(stdinput)
	if results:
	for result in results:
	print result


	if __name__ == "__main__":
	main()