-
Notifications
You must be signed in to change notification settings - Fork 0
/
measure_resource_usage.py
100 lines (76 loc) · 3.24 KB
/
measure_resource_usage.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
import random
import resource
from turnout_election_schemes.schemes.majorityjudgement import VoteAggregator, MajorityJudgementCount
from turnout_election_schemes.schemes.singletransferablevote import SingleTransferableVoteScheme
import sys
class MajorityJudgementDataGenerator(object):
def __init__(self, num_grades, num_candidates, num_voters):
self.num_grades = num_grades
self.num_candidates = num_candidates
self.num_voters = num_voters
def votes(self):
return tuple(self.vote() for _ in range(self.num_voters))
def vote(self):
return tuple(self.grade() for candidate in self.candidates())
def grade(self):
return random.randint(0, self.num_grades - 1)
def candidates(self):
return range(self.num_candidates)
class SingleTransferableVoteDataGenerator(object):
def __init__(self, num_candidates, num_voters):
self.num_candidates = num_candidates
self.num_voters = num_voters
def votes(self):
return tuple(self.vote() for _ in range(self.num_voters))
def vote(self):
return random.sample(self.candidates(), random.randrange(len(self.candidates()) + 1))
def candidates(self):
return range(self.num_candidates)
def usage():
print 'Usage: %s mj num_grades num_candidates num_voters' % sys.argv[0]
print ' %s stv num_vacancies num_candidates num_voters' % sys.argv[0]
def resource_usage():
return resource.getrusage(resource.RUSAGE_SELF)
def print_resource_usage(initial, final):
utime = final.ru_utime - initial.ru_utime
stime = final.ru_stime - initial.ru_stime
maxrss = final.ru_maxrss - initial.ru_maxrss
print 'utime: %f\t\tstime: %f\t\tmaxrss: %i' % (utime, stime, maxrss)
def test_majority_judgement(num_grades, num_candidates, num_voters):
generator = MajorityJudgementDataGenerator(num_grades, num_candidates, num_voters)
candidates = generator.candidates()
votes = generator.votes()
initial_resource_usage = resource_usage()
aggregator = VoteAggregator(candidates, generator.num_grades)
aggregated_votes = aggregator.aggregate(votes)
scheme = MajorityJudgementCount()
scheme.sort_candidates(aggregated_votes)
print_resource_usage(initial_resource_usage, resource_usage())
def test_single_transferable_vote(num_vacancies, num_candidates, num_voters):
generator = SingleTransferableVoteDataGenerator(num_candidates, num_voters)
candidates = generator.candidates()
votes = generator.votes()
initial_resource_usage = resource_usage()
scheme = SingleTransferableVoteScheme(num_vacancies, candidates, votes, random = random)
rounds = 0
while not scheme.completed():
scheme.run_round()
rounds += 1
if rounds > num_candidates:
print 'Election did not complete within %d rounds' % rounds
break
print_resource_usage(initial_resource_usage, resource_usage())
if __name__ == '__main__':
if len(sys.argv) != 5:
usage()
sys.exit(1)
try:
system = sys.argv[1]
args = map(int, sys.argv[2:])
except ValueError:
usage()
sys.exit(1)
if system == 'mj':
test_majority_judgement(*args)
elif system == 'stv':
test_single_transferable_vote(*args)