This repository has been archived on 2021-05-25. You can view files and clone it, but cannot push or open issues or pull requests.
pyRCV2/pyRCV2/method/base_stv.py

549 lines
17 KiB
Python
Raw Normal View History

2020-10-17 22:20:13 +11:00
# pyRCV2: Preferential vote counting
# Copyright © 2020 Lee Yingtong Li (RunasSudo)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
__pragma__ = lambda x: None
from pyRCV2.model import CandidateState, CountCard, CountCompleted, CountStepResult
from pyRCV2.numbers import Num, Rational
2020-10-17 22:20:13 +11:00
from pyRCV2.safedict import SafeDict
class STVException(Exception):
pass
class BaseSTVCounter:
"""
Basic STV counter for various different variations
"""
2020-10-17 22:20:13 +11:00
def __init__(self, election, options=None):
2020-10-17 22:20:13 +11:00
self.election = election
self.cls_ballot_value = Num # Need to use Rational in unweighted inclusive Gregory
# Default options
self.options = {
'prog_quota': False, # Progressively reducing quota?
'bulk_elect': True, # Bulk election?
'quota': 'droop', # 'droop', 'droop_exact', 'hare' or 'hare_exact'
'quota_criterion': 'geq', # 'geq' or 'gt'
'surplus_order': 'size', # 'size' or 'order'
2020-12-24 00:04:30 +11:00
'ties': []
}
if options is not None:
self.options.update(options)
2020-10-18 21:24:12 +11:00
2020-10-17 22:20:13 +11:00
self.candidates = SafeDict([(c, CountCard()) for c in self.election.candidates])
self.exhausted = CountCard()
2020-10-18 03:48:00 +11:00
self.loss_fraction = CountCard()
self.total_orig = sum((b.value for b in self.election.ballots), Num('0'))
2020-10-17 22:20:13 +11:00
self.num_elected = 0
2020-10-17 22:20:13 +11:00
# Withdraw candidates
for candidate in self.election.withdrawn:
__pragma__('opov')
self.candidates[candidate].state = CandidateState.WITHDRAWN
__pragma__('noopov')
2020-10-18 21:24:12 +11:00
def reset(self):
"""
Public function:
2020-12-24 01:36:39 +11:00
Perform the first step (distribute first preferences)
Does not reset the states of candidates, etc.
2020-10-18 21:24:12 +11:00
"""
# Distribute first preferences
for ballot in self.election.ballots:
__pragma__('opov')
candidate = next((c for c in ballot.preferences if self.candidates[c].state == CandidateState.HOPEFUL), None)
if candidate is not None:
self.candidates[candidate].transfers += ballot.value
self.candidates[candidate].ballots.append((ballot, self.cls_ballot_value(ballot.value)))
else:
self.exhausted.transfers += ballot.value
self.exhausted.ballots.append((ballot, self.cls_ballot_value(ballot.value)))
__pragma__('noopov')
self.quota = None
self.compute_quota()
self.elect_meeting_quota()
__pragma__('opov')
2020-12-24 01:36:39 +11:00
result = CountStepResult(
'First preferences',
self.candidates,
self.exhausted,
self.loss_fraction,
self.total + self.exhausted.votes + self.loss_fraction.votes,
self.quota
)
__pragma__('noopov')
2020-12-24 01:36:39 +11:00
self.step_results = [result]
return result
2020-10-18 21:24:12 +11:00
2020-10-17 22:20:13 +11:00
def step(self):
2020-10-18 21:24:12 +11:00
"""
Public function:
Perform one step of the STV count
"""
2020-10-17 22:20:13 +11:00
# Step count cards
2020-10-18 21:24:12 +11:00
self.step_count_cards()
# Check if done
result = self.before_surpluses()
2020-10-18 21:24:12 +11:00
if result:
return result
# Distribute surpluses
result = self.distribute_surpluses()
if result:
return result
# Check if done (2)
result = self.before_exclusion()
if result:
return result
2020-10-18 21:24:12 +11:00
# Insufficient winners and no surpluses to distribute
# Exclude the lowest ranked hopeful
result = self.exclude_candidate()
if result:
return result
raise STVException('Unable to complete step')
def step_count_cards(self):
"""
Reset the count cards for the beginning of a new step
"""
2020-10-17 22:20:13 +11:00
for candidate, count_card in self.candidates.items():
count_card.step()
self.exhausted.step()
2020-10-18 03:48:00 +11:00
self.loss_fraction.step()
2020-10-18 21:24:12 +11:00
def before_surpluses(self):
2020-10-18 21:24:12 +11:00
"""
Check if the count can be completed before distributing surpluses
2020-10-18 21:24:12 +11:00
"""
2020-10-17 22:20:13 +11:00
# Have sufficient candidates been elected?
if self.num_elected >= self.election.seats:
2020-10-17 22:20:13 +11:00
return CountCompleted()
# Are there just enough candidates to fill all the seats?
if self.options['bulk_elect']:
if self.num_elected + sum(1 for c, cc in self.candidates.items() if cc.state == CandidateState.HOPEFUL) <= self.election.seats:
# TODO: Sort by size
# Declare elected all remaining candidates
for candidate, count_card in self.candidates.items():
if count_card.state == CandidateState.HOPEFUL:
count_card.state = CandidateState.PROVISIONALLY_ELECTED
count_card.order_elected = self.num_elected
self.num_elected += 1
__pragma__('opov')
result = CountStepResult(
'Bulk election',
self.candidates,
self.exhausted,
self.loss_fraction,
self.total + self.exhausted.votes + self.loss_fraction.votes,
self.quota
)
__pragma__('noopov')
self.step_results.append(result)
return result
2020-10-18 21:24:12 +11:00
def distribute_surpluses(self):
"""
Distribute surpluses, if any
"""
2020-10-17 22:20:13 +11:00
# Do surpluses need to be distributed?
__pragma__('opov')
has_surplus = [(c, cc) for c, cc in self.candidates.items() if cc.state == CandidateState.PROVISIONALLY_ELECTED and cc.votes > self.quota]
__pragma__('noopov')
if len(has_surplus) > 0:
# Distribute surpluses in specified order
if self.options['surplus_order'] == 'size':
2020-12-23 21:39:43 +11:00
has_surplus.sort(key=lambda x: x[1].votes, reverse=True)
2020-12-23 22:36:49 +11:00
candidate_surplus, count_card = self.choose_highest(has_surplus) # May need to break ties
elif self.options['surplus_order'] == 'order':
2020-12-23 21:39:43 +11:00
has_surplus.sort(key=lambda x: x[1].order_elected)
2020-12-23 22:36:49 +11:00
candidate_surplus, count_card = has_surplus[0] # Ties were already broken when these were assigned
else:
raise STVException('Invalid surplus order option')
2020-10-17 22:20:13 +11:00
count_card.state = CandidateState.ELECTED
__pragma__('opov')
surplus = count_card.votes - self.quota
__pragma__('noopov')
# Transfer surplus
2020-10-18 21:24:12 +11:00
self.do_surplus(candidate_surplus, count_card, surplus)
2020-10-17 22:20:13 +11:00
2020-10-18 03:25:41 +11:00
__pragma__('opov')
count_card.transfers -= surplus
__pragma__('noopov')
2020-10-18 21:24:12 +11:00
# Declare elected any candidates meeting the quota as a result of surpluses
2020-10-17 22:20:13 +11:00
self.compute_quota()
self.elect_meeting_quota()
2020-10-18 02:54:51 +11:00
__pragma__('opov')
2020-12-24 01:36:39 +11:00
result = CountStepResult(
2020-10-17 23:09:29 +11:00
'Surplus of ' + candidate_surplus.name,
2020-10-17 22:20:13 +11:00
self.candidates,
self.exhausted,
2020-10-18 03:48:00 +11:00
self.loss_fraction,
self.total + self.exhausted.votes + self.loss_fraction.votes,
2020-10-17 22:20:13 +11:00
self.quota
)
2020-10-18 02:54:51 +11:00
__pragma__('noopov')
2020-12-24 01:36:39 +11:00
self.step_results.append(result)
return result
2020-10-18 21:24:12 +11:00
def do_surplus(self, candidate_surplus, count_card, surplus):
"""
Transfer the surplus of the given candidate
Subclasses must override this function
"""
raise NotImplementedError('Method not implemented')
def before_exclusion(self):
"""
Check before excluding a candidate
"""
# If we did not perform bulk election in before_surpluses: Are there just enough candidates to fill all the seats?
if not self.options['bulk_elect']:
if self.num_elected + sum(1 for c, cc in self.candidates.items() if cc.state == CandidateState.HOPEFUL) <= self.election.seats:
# Declare elected one remaining candidate at a time
hopefuls = [(c, cc) for c, cc in self.candidates.items() if cc.state == CandidateState.HOPEFUL]
hopefuls.sort(key=lambda x: x[1].votes, reverse=True)
candidate_elected, count_card = self.choose_highest(hopefuls)
count_card.state = CandidateState.PROVISIONALLY_ELECTED
count_card.order_elected = self.num_elected
self.num_elected += 1
__pragma__('opov')
result = CountStepResult(
'Bulk election',
self.candidates,
self.exhausted,
self.loss_fraction,
self.total + self.exhausted.votes + self.loss_fraction.votes,
self.quota
)
__pragma__('noopov')
self.step_results.append(result)
return result
2020-10-18 21:24:12 +11:00
def exclude_candidate(self):
"""
Exclude the lowest ranked hopeful
"""
2020-10-17 22:20:13 +11:00
2020-10-18 21:47:59 +11:00
candidate_excluded, count_card = self.candidate_to_exclude()
2020-10-17 22:20:13 +11:00
count_card.state = CandidateState.EXCLUDED
# Exclude this candidate
2020-10-18 21:24:12 +11:00
self.do_exclusion(candidate_excluded, count_card)
2020-10-17 22:20:13 +11:00
__pragma__('opov')
count_card.transfers -= count_card.votes
__pragma__('noopov')
# Declare any candidates meeting the quota as a result of exclusion
self.compute_quota()
self.elect_meeting_quota()
2020-10-18 02:54:51 +11:00
__pragma__('opov')
2020-12-24 01:36:39 +11:00
result = CountStepResult(
2020-10-17 23:09:29 +11:00
'Exclusion of ' + candidate_excluded.name,
2020-10-17 22:20:13 +11:00
self.candidates,
self.exhausted,
2020-10-18 03:48:00 +11:00
self.loss_fraction,
self.total + self.exhausted.votes + self.loss_fraction.votes,
2020-10-17 22:20:13 +11:00
self.quota
)
2020-10-18 02:54:51 +11:00
__pragma__('noopov')
2020-12-24 01:36:39 +11:00
self.step_results.append(result)
return result
2020-10-17 22:20:13 +11:00
2020-10-18 21:47:59 +11:00
def candidate_to_exclude(self):
"""
Determine the candidate to exclude
"""
hopefuls = [(c, cc) for c, cc in self.candidates.items() if cc.state == CandidateState.HOPEFUL]
2020-12-23 21:39:43 +11:00
hopefuls.sort(key=lambda x: x[1].votes)
2020-10-18 21:47:59 +11:00
2020-12-23 22:36:49 +11:00
candidate_excluded, count_card = self.choose_lowest(hopefuls)
2020-10-18 21:47:59 +11:00
return candidate_excluded, count_card
2020-10-18 21:24:12 +11:00
def do_exclusion(self, candidate_excluded, count_card):
"""
Exclude the given candidate and transfer the votes
Subclasses must override this function
"""
raise NotImplementedError('Method not implemented')
2020-10-17 22:20:13 +11:00
def compute_quota(self):
2020-10-18 21:24:12 +11:00
"""
Recount total votes and (if applicable) recalculate the quota
"""
2020-10-17 22:20:13 +11:00
__pragma__('opov')
2020-10-18 02:54:51 +11:00
self.total = sum((cc.votes for c, cc in self.candidates.items()), Num('0'))
2020-10-18 03:48:00 +11:00
self.loss_fraction.transfers += (self.total_orig - self.total - self.exhausted.votes) - self.loss_fraction.votes
if self.quota is None or self.options['prog_quota']:
if self.options['quota'] == 'droop':
self.quota = (self.total / Num(self.election.seats + 1)).__floor__() + Num('1')
elif self.options['quota'] == 'droop_exact':
self.quota = self.total / Num(self.election.seats + 1)
elif self.options['quota'] == 'hare':
self.quota = (self.total / Num(self.election.seats)).__floor__() + Num('1')
elif self.options['quota'] == 'hare_exact':
self.quota = self.total / Num(self.election.seats)
else:
raise STVException('Invalid quota option')
2020-10-17 22:20:13 +11:00
__pragma__('noopov')
def meets_quota(self, count_card):
2020-10-18 21:24:12 +11:00
"""
Determine if the given candidate meets the quota
"""
if self.options['quota_criterion'] == 'geq':
__pragma__('opov')
return count_card.votes >= self.quota
__pragma__('noopov')
elif self.options['quota_criterion'] == 'gt':
__pragma__('opov')
return count_card.votes > self.quota
__pragma__('noopov')
else:
raise STVException('Invalid quota criterion')
2020-10-17 22:20:13 +11:00
def elect_meeting_quota(self):
2020-10-18 21:24:12 +11:00
"""
Elect all candidates meeting the quota
"""
2020-10-17 22:20:13 +11:00
# Does a candidate meet the quota?
meets_quota = [(c, cc) for c, cc in self.candidates.items() if cc.state == CandidateState.HOPEFUL and self.meets_quota(cc)]
2020-10-17 22:20:13 +11:00
if len(meets_quota) > 0:
2020-12-23 22:36:49 +11:00
meets_quota.sort(key=lambda x: x[1].votes, reverse=True)
2020-10-17 22:20:13 +11:00
# Declare elected any candidate who meets the quota
2020-12-23 22:36:49 +11:00
while len(meets_quota) > 0:
x = self.choose_highest(meets_quota)
candidate, count_card = x[0], x[1]
2020-10-17 22:20:13 +11:00
count_card.state = CandidateState.PROVISIONALLY_ELECTED
count_card.order_elected = self.num_elected
self.num_elected += 1
2020-12-23 22:36:49 +11:00
meets_quota.remove(x)
def choose_lowest(self, l):
"""
Provided a list of tuples (Candidate, CountCard), sorted in ASCENDING order of votes, choose the tuple with the fewest votes, breaking ties appropriately
"""
if len(l) == 1:
return l[0]
2020-12-24 00:04:30 +11:00
__pragma__('opov')
# Do not use (c, cc) for c, cc in ... as this will break equality in JS
tied = [x for x in l if x[1].votes == l[0][1].votes]
__pragma__('noopov')
2020-12-23 22:36:49 +11:00
if len(tied) == 1:
return tied[0]
# A tie exists
for tie in self.options['ties']:
result = tie.choose_lowest(tied)
if result is not None:
return result
raise Exception('Unable to resolve tie')
def choose_highest(self, l):
"""
Provided a list of tuples (Candidate, CountCard), sorted in DESCENDING order of votes, choose the tuple with the most votes, breaking ties appropriately
"""
if len(l) == 1:
return l[0]
2020-12-24 00:04:30 +11:00
__pragma__('opov')
# Do not use (c, cc) for c, cc in ... as this will break equality in JS
tied = [x for x in l if x[1].votes == l[0][1].votes]
__pragma__('noopov')
2020-12-23 22:36:49 +11:00
if len(tied) == 1:
return tied[0]
# A tie exists
for tie in self.options['ties']:
result = tie.choose_highest(tied)
if result is not None:
return result
raise Exception('Unable to resolve tie')
class BaseWIGSTVCounter(BaseSTVCounter):
"""
Basic weighted inclusive Gregory STV counter
"""
2020-10-18 21:24:12 +11:00
def do_surplus(self, candidate_surplus, count_card, surplus):
for ballot, ballot_value in count_card.ballots:
__pragma__('opov')
new_value = (ballot_value * surplus) / count_card.votes # Multiply first to avoid rounding errors
candidate = next((c for c in ballot.preferences if self.candidates[c].state == CandidateState.HOPEFUL), None)
if candidate is not None:
self.candidates[candidate].transfers += new_value
self.candidates[candidate].ballots.append((ballot, new_value))
else:
self.exhausted.transfers += new_value
self.exhausted.ballots.append((ballot, new_value))
__pragma__('noopov')
2020-10-18 21:24:12 +11:00
def do_exclusion(self, candidate_excluded, count_card):
for ballot, ballot_value in count_card.ballots:
__pragma__('opov')
candidate = next((c for c in ballot.preferences if self.candidates[c].state == CandidateState.HOPEFUL), None)
if candidate is not None:
self.candidates[candidate].transfers += ballot_value
self.candidates[candidate].ballots.append((ballot, ballot_value))
else:
self.exhausted.transfers += ballot_value
self.exhausted.ballots.append((ballot, ballot_value))
__pragma__('noopov')
class BaseUIGSTVCounter(BaseSTVCounter):
"""
Basic unweighted inclusive Gregory STV counter
"""
def __init__(self, *args, **kwargs):
BaseSTVCounter.__init__(self, *args, **kwargs)
# Need to use Rational for ballot value internally, as Num may be set to integers only
self.cls_ballot_value = Rational
2020-10-18 21:24:12 +11:00
def do_surplus(self, candidate_surplus, count_card, surplus):
# FIXME: Is it okay to use native int's here?
next_preferences = SafeDict([(c, []) for c, cc in self.candidates.items()])
next_exhausted = []
total_ballots = Num('0')
# Count next preferences
for ballot, ballot_value in count_card.ballots:
__pragma__('opov')
total_ballots += ballot.value
candidate = next((c for c in ballot.preferences if self.candidates[c].state == CandidateState.HOPEFUL), None)
__pragma__('noopov')
if candidate is not None:
__pragma__('opov')
next_preferences[candidate].append(ballot)
__pragma__('noopov')
else:
next_exhausted.append(ballot)
# Make transfers
for candidate, cand_ballots in next_preferences.items():
num_ballots = sum((b.value for b in cand_ballots), Num('0'))
__pragma__('opov')
self.candidates[candidate].transfers += (num_ballots * surplus) / total_ballots # Multiply first to avoid rounding errors
__pragma__('noopov')
for ballot in cand_ballots:
__pragma__('opov')
new_value = (ballot.value.to_rational() * surplus.to_rational()) / total_ballots.to_rational()
self.candidates[candidate].ballots.append((ballot, new_value))
__pragma__('noopov')
num_exhausted = sum((b.value for b in next_exhausted), Num('0'))
__pragma__('opov')
self.exhausted.transfers += (num_exhausted * surplus) / total_ballots
__pragma__('noopov')
for ballot in next_exhausted:
__pragma__('opov')
new_value = (ballot.value.to_rational() * surplus.to_rational()) / total_ballots.to_rational()
__pragma__('noopov')
self.exhausted.ballots.append((ballot, new_value))
2020-10-18 21:24:12 +11:00
def do_exclusion(self, candidate_excluded, count_card):
next_preferences = SafeDict([(c, Rational('0')) for c, cc in self.candidates.items()])
next_exhausted = Rational('0')
total_votes = Rational('0')
# Count and transfer next preferences
for ballot, ballot_value in count_card.ballots:
__pragma__('opov')
total_votes += ballot_value
candidate = next((c for c in ballot.preferences if self.candidates[c].state == CandidateState.HOPEFUL), None)
__pragma__('noopov')
if candidate is not None:
__pragma__('opov')
next_preferences[candidate] += ballot_value
self.candidates[candidate].ballots.append((ballot, ballot_value))
__pragma__('noopov')
else:
__pragma__('opov')
next_exhausted += ballot_value
__pragma__('noopov')
self.exhausted.ballots.append((ballot, ballot_value))
# Credit votes
for candidate, cand_votes in next_preferences.items():
__pragma__('opov')
self.candidates[candidate].transfers += cand_votes.to_num()
__pragma__('noopov')
__pragma__('opov')
self.exhausted.transfers += next_exhausted.to_num()
__pragma__('noopov')