from __future__ import annotations
import datetime
from collections.abc import Mapping
from onegov.core.orm import Base, observes
from onegov.core.orm import translation_hybrid
from onegov.core.orm.mixins import ContentMixin
from onegov.core.orm.mixins import dict_property
from onegov.core.orm.mixins import meta_property
from onegov.core.orm.types import HSTORE
from onegov.election_day.models.election.candidate import Candidate
from onegov.election_day.models.election.election_result import ElectionResult
from onegov.election_day.models.election.mixins import DerivedAttributesMixin
from onegov.election_day.models.mixins import DomainOfInfluenceMixin
from onegov.election_day.models.mixins import ExplanationsPdfMixin
from onegov.election_day.models.mixins import IdFromTitlesMixin
from onegov.election_day.models.mixins import LastModifiedMixin
from onegov.election_day.models.mixins import StatusMixin
from onegov.election_day.models.mixins import summarized_property
from onegov.election_day.models.mixins import TitleTranslationsMixin
from onegov.election_day.models.party_result.mixins import (
PartyResultsOptionsMixin)
from operator import itemgetter
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import select
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import object_session
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DynamicMapped
from sqlalchemy.orm import Mapped
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from onegov.election_day.models import DataSourceItem
from onegov.election_day.models import ElectionCompound
from onegov.election_day.models import ElectionRelationship
from onegov.election_day.models import Notification
from onegov.election_day.models import Screen
from sqlalchemy.orm import Query
from sqlalchemy.sql import ColumnElement
from typing import NamedTuple
[docs]
class VotesByDistrictRow(NamedTuple):
[docs]
class Election(Base, ContentMixin, LastModifiedMixin,
DomainOfInfluenceMixin, StatusMixin, TitleTranslationsMixin,
IdFromTitlesMixin, DerivedAttributesMixin, ExplanationsPdfMixin,
PartyResultsOptionsMixin):
[docs]
__tablename__ = 'elections'
@property
[docs]
def polymorphic_base(self) -> type[Election]:
return Election
#: the type of the item, this can be used to create custom polymorphic
#: subclasses of this class. See
#: `<https://docs.sqlalchemy.org/en/improve_toc/\
#: orm/extensions/declarative/inheritance.html>`_.
[docs]
type: Mapped[str] = mapped_column()
[docs]
__mapper_args__ = {
'polymorphic_on': type,
'polymorphic_identity': 'majorz'
}
#: Identifies the election, may be used in the url
[docs]
id: Mapped[str] = mapped_column(primary_key=True)
#: external identifier
[docs]
external_id: Mapped[str | None]
#: all translations of the title
[docs]
title_translations: Mapped[Mapping[str, str]] = mapped_column(HSTORE)
#: the translated title (uses the locale of the request, falls back to the
#: default locale of the app)
[docs]
title = translation_hybrid(title_translations)
#: all translations of the short title
[docs]
short_title_translations: Mapped[Mapping[str, str] | None] = mapped_column(
HSTORE
)
#: the translated short title (uses the locale of the request, falls back
#: to the default locale of the app)
[docs]
short_title = translation_hybrid(short_title_translations)
@observes('title_translations', 'short_title_translations')
[docs]
def title_observer(
self,
title_translations: Mapping[str, str],
short_title_translations: Mapping[str, str]
) -> None:
if not self.id:
session = object_session(self)
assert session is not None
self.id = self.id_from_title(session)
#: Shortcode for cantons that use it
[docs]
shortcode: Mapped[str | None]
#: The date of the election
[docs]
date: Mapped[datetime.date]
#: Number of mandates
[docs]
number_of_mandates: Mapped[int] = mapped_column(default=lambda: 0)
@property
[docs]
def allocated_mandates(self) -> int:
""" Number of already allocated mandates/elected candidates. """
# Unless an election is finished, allocated mandates are 0
if not self.completed:
return 0
return sum(c.elected for c in self.candidates)
#: Defines the type of majority (e.g. 'absolute', 'relative')
[docs]
majority_type: dict_property[str | None] = meta_property('majority_type')
#: Absolute majority
[docs]
absolute_majority: Mapped[int | None]
@hybrid_property
[docs]
def counted(self) -> bool:
""" True if all results have been counted. """
if not self.results:
return False
return all(r.counted for r in self.results)
@counted.inplace.expression
@classmethod
[docs]
def _counted_expression(cls) -> ColumnElement[bool]:
expr = select(
func.coalesce(func.bool_and(ElectionResult.counted), False)
)
expr = expr.where(ElectionResult.election_id == cls.id)
return expr.label('counted')
@property
[docs]
def progress(self) -> tuple[int, int]:
""" Returns a tuple with the first value being the number of counted
election results and the second value being the number of total
results.
"""
return sum(r.counted for r in self.results), len(self.results)
@property
[docs]
def counted_entities(self) -> list[str]:
""" Returns the names of the already counted entities.
Might contain an empty string in case of expats.
"""
return sorted(r.name for r in self.results if r.counted)
@property
[docs]
def has_results(self) -> bool:
""" Returns True, if the election has any results. """
for result in self.results:
if result.counted:
return True
return False
#: An election contains n candidates
[docs]
candidates: Mapped[list[Candidate]] = relationship(
cascade='all, delete-orphan',
back_populates='election',
order_by='Candidate.candidate_id',
)
#: An election contains n results, one for each political entity
[docs]
results: Mapped[list[ElectionResult]] = relationship(
cascade='all, delete-orphan',
back_populates='election',
order_by='ElectionResult.district, ElectionResult.name',
)
@property
[docs]
def results_query(self) -> Query[ElectionResult]:
session = object_session(self)
assert session is not None
query = session.query(ElectionResult)
query = query.filter(ElectionResult.election_id == self.id)
query = query.order_by(ElectionResult.district, ElectionResult.name)
return query
#: An election may have related elections
#: An election may be related by other elections
[docs]
referencing_elections: DynamicMapped[ElectionRelationship] = relationship(
foreign_keys='ElectionRelationship.target_id',
cascade='all, delete-orphan',
back_populates='target',
)
#: An election may be part of an election compound
[docs]
election_compound_id: Mapped[str | None] = mapped_column(
ForeignKey('election_compounds.id', onupdate='CASCADE')
)
#: The election compound this election belongs to
[docs]
election_compound: Mapped[ElectionCompound | None] = relationship(
back_populates='elections'
)
@property
[docs]
def completed(self) -> bool:
""" Overwrites StatusMixin's 'completed' for compounds with manual
completion. """
result = super().completed
compound = self.election_compound
if compound and compound.completes_manually:
return compound.manually_completed and result
return result
#: The total eligible voters
[docs]
eligible_voters = summarized_property('eligible_voters')
#: The expats
[docs]
expats = summarized_property('expats')
#: The total received ballots
[docs]
received_ballots = summarized_property('received_ballots')
#: The total accounted ballots
[docs]
accounted_ballots = summarized_property('accounted_ballots')
#: The total blank ballots
[docs]
blank_ballots = summarized_property('blank_ballots')
#: The total invalid ballots
[docs]
invalid_ballots = summarized_property('invalid_ballots')
#: The total accounted votes
[docs]
accounted_votes = summarized_property('accounted_votes')
[docs]
def aggregate_results(self, attribute: str) -> int:
""" Gets the sum of the given attribute from the results. """
return sum(getattr(result, attribute) or 0 for result in self.results)
@classmethod
[docs]
def aggregate_results_expression(
cls,
attribute: str
) -> ColumnElement[int]:
""" Gets the sum of the given attribute from the results,
as SQL expression.
"""
expr = select(
func.coalesce(
func.sum(getattr(ElectionResult, attribute)),
0
)
)
expr = expr.where(ElectionResult.election_id == cls.id)
return expr.label(attribute)
@property
[docs]
def elected_candidates(self) -> list[tuple[str, str]]:
""" Returns the first and last names of the elected candidates. """
return sorted(
(
(c.first_name, c.family_name)
for c in self.candidates if c.elected
),
key=itemgetter(1, 0)
)
#: may be used to store a link related to this election
#: may be used to mark an election as a tacit election
[docs]
tacit: dict_property[bool] = meta_property('tacit', default=False)
#: may be used to indicate that the vote contains expats as seperate
#: results (typically with entity_id = 0)
[docs]
has_expats: dict_property[bool] = meta_property('expats', default=False)
#: The segment of the domain. This might be the district, if this is a
#: regional (district) election; the region, if it's a regional (region)
#: election or the municipality, if this is a communal election.
[docs]
domain_segment: dict_property[str] = meta_property(
'domain_segment',
default=''
)
#: The supersegment of the domain. This might be superregion, if it's a
#: regional (region) election.
[docs]
domain_supersegment: dict_property[str] = meta_property(
'domain_supersegment',
default=''
)
@property
[docs]
def votes_by_district(self) -> Query[VotesByDistrictRow]:
query = self.results_query.order_by(None)
results: Query[VotesByDistrictRow] = query.with_entities( # type: ignore[assignment]
self.__class__.id.label('election_id'),
ElectionResult.district,
func.array_agg(
ElectionResult.entity_id.distinct()).label('entities'),
func.coalesce(
func.bool_and(ElectionResult.counted), False
).label('counted'),
func.coalesce(
func.sum(ElectionResult.accounted_ballots), 0).label('votes')
)
results = results.group_by(
ElectionResult.district,
self.__class__.id.label('election_id')
)
return results
#: Defines optional colors for lists and parties
[docs]
colors: dict_property[dict[str, str]] = meta_property(
'colors',
default=dict
)
[docs]
def clear_results(self, clear_all: bool = False) -> None:
""" Clears all the results. """
self.absolute_majority = None
self.status = None
self.last_result_change = None
session = object_session(self)
assert session is not None
if clear_all:
session.query(Candidate).filter(
Candidate.election_id == self.id
).delete()
session.query(ElectionResult).filter(
ElectionResult.election_id == self.id
).delete()
session.flush()
session.expire_all()
#: data source items linked to this election
[docs]
data_sources: Mapped[list[DataSourceItem]] = relationship(
back_populates='election'
)
#: notifcations linked to this election
[docs]
notifications: DynamicMapped[Notification] = relationship(
'onegov.election_day.models.notification.Notification',
back_populates='election',
)
#: screens linked to this election
[docs]
screens: DynamicMapped[Screen] = relationship(
back_populates='election',
)