10794283251

Committed 10 Sep 2024 02:11PM UTC coverage: 85.85% (-1.6%) from 87.44%

Build # 10794283251

Build Type

Pull #1343

github

Committed by

web-flow

Commit Message

Merge f0ded7995 into 1e2dbc28e

Pull Request Pull Request #1343: Add complex shift structure

Run Details

2920 of 3393 branches covered (86.06%)

Branch coverage included in aggregate %.

195 of 433 new or added lines in 17 files covered. (45.03%)

5 existing lines in 3 files now uncovered.

11538 of 13448 relevant lines covered (85.8%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.61

/ephios/core/services/matching.py

import dataclasses
from typing import Collection, Optional

from django.utils.functional import cached_property
from scipy.sparse import csr_matrix
from scipy.sparse.csgraph import min_weight_full_bipartite_matching

from ephios.core.models import Qualification
from ephios.core.services.qualification import QualificationUniverse
from ephios.core.signup.participants import AbstractParticipant


@dataclasses.dataclass(unsafe_hash=True)
class Position:
    id: str
    required: bool
    required_qualifications: Collection[Qualification]
    designated_for: Collection[AbstractParticipant]  # designated by disposition
    preferred_by: Collection[AbstractParticipant]  # preferred by participant (less important)
    label: Optional[str] = None
    aux_score: float = 0.0  # additional score control in range [0,1]

    def __post_init__(self):
        self.required_qualifications = frozenset(self.required_qualifications)
        self.preferred_by = frozenset(self.preferred_by)
        self.designated_for = frozenset(self.designated_for)

    @cached_property
    def required_skill(self):
        graph = QualificationUniverse.get_graph()
        required_qualification_uuids = [
            qualification.uuid for qualification in self.required_qualifications
        ]
        return set(graph.spread_from(required_qualification_uuids))

    @cached_property
    def skill_level(self):
        return skill_level(self.required_skill)


def skill_level(qualifications):
    """
    skill_level is some float in the range [0,1] that expresses
    how far up in the qualification hierarchy the required qualifications are.
    """
    if not qualifications:
        return 0
    qualifications = list(qualifications)
    # convert to uuids
    if isinstance(qualifications[0], Qualification):
        qualifications = [q.uuid for q in qualifications]
    graph = QualificationUniverse.get_graph()
    required_skill = set(graph.spread_from(qualifications))
    # all_skill is qualifications up and down from the required ones
    all_skill = required_skill | set(graph.spread_reverse(qualifications))
    return len(required_skill) / len(all_skill)


class Matching:

    def __init__(self, participants, positions, pairings):
        self.pairings = pairings
        self.participants = participants
        self.positions = positions
        self.unpaired_participants = set(participants)
        self.unpaired_positions = list(
            sorted(positions, key=lambda position: (-position.skill_level, position.id))
        )
        for participant, position in pairings:
            self.unpaired_participants.remove(participant)
            self.unpaired_positions.remove(position)
        self._participations = None

    @property
    def participations(self):
        if self._participations is None:
            raise ValueError("participations not attached")
        return self._participations

    def attach_participations(self, participations):
        self._participations = participations

    @cached_property
    def _participation_by_participant(self):
        return {p.participant: p for p in self.participations}

    @cached_property
    def participation_pairings(self):
        return [
            (self._participation_by_participant[participant], position)
            for participant, position in self.pairings
        ]

    @cached_property
    def unpaired_participations(self):
        return [
            participation
            for participation in self.participations
            if participation.participant in self.unpaired_participants
        ]

    def participation_for_position(self, position_id):
        for participation, position in self.participation_pairings:
            if position.id == position_id:
                return participation


def score_pairing(
    participant: AbstractParticipant, position: Position, number_of_participants=1_000_000
):
    """
    Score the pairing of participant and  position.
    Skill here means a set of qualification.
    """
    # number of participants is used as a scoring value to make sure that in adverse cases
    # a matching with a someone unqualified pairings gets a worse score than a matching with more valid pairings
    # (similar for prefers and skill level). It can be arbitrarily big, but must be at least bigger than the sum of
    # all the other constants used in the score.
    padded_participant_count = 10 + 2 * number_of_participants
    base_score = 1.0
    preferred_value = 3.0
    max_skill_value = 1.0
    max_aux_value = 2.0
    required_value = padded_participant_count * sum(
        (base_score, preferred_value, max_skill_value, max_aux_value)
    )
    designated_value = required_value**2
    unqualified_penalty = designated_value**2

    is_designated = participant in position.designated_for

    if not is_designated and not position.required_skill <= participant.skill:
        # the participant does not have some required skill
        return -unqualified_penalty  # avoid matching unqualified participants

    score = base_score
    if is_designated:
        score += designated_value
    if participant in position.preferred_by:
        score += preferred_value
    if position.required:
        score += required_value
    score += position.skill_level * max_skill_value
    score += position.aux_score * max_aux_value
    return score


def match_participants_to_positions(
    participants: Collection[AbstractParticipant],
    positions: Collection[Position],
) -> Matching:
    participants = list(participants)
    positions = list(positions)
    costs = csr_matrix(
        [
            [
                -score_pairing(participant, position, number_of_participants=len(participants))
                for position in positions
            ]
            for participant in participants
        ]
    )
    matching = min_weight_full_bipartite_matching(costs)
    pairings = set()
    for par_idx, pos_idx in zip(*matching):
        cost = costs[par_idx, pos_idx]
        if cost <= 0:
            pairings.add((participants[par_idx], positions[pos_idx]))
    return Matching(participants, positions, pairings)

1	import dataclasses	1✔
2	from typing import Collection, Optional	1✔
3
4	from django.utils.functional import cached_property	1✔
5	from scipy.sparse import csr_matrix	1✔
6	from scipy.sparse.csgraph import min_weight_full_bipartite_matching	1✔
7
8	from ephios.core.models import Qualification	1✔
9	from ephios.core.services.qualification import QualificationUniverse	1✔
10	from ephios.core.signup.participants import AbstractParticipant	1✔
11
12
13	@dataclasses.dataclass(unsafe_hash=True)	1✔
14	class Position:	1✔
15	id: str	1✔
16	required: bool	1✔
17	required_qualifications: Collection[Qualification]	1✔
18	designated_for: Collection[AbstractParticipant] # designated by disposition	1✔
19	preferred_by: Collection[AbstractParticipant] # preferred by participant (less important)	1✔
20	label: Optional[str] = None	1✔
21	aux_score: float = 0.0 # additional score control in range [0,1]	1✔
22
23	def __post_init__(self):	1✔
24	self.required_qualifications = frozenset(self.required_qualifications)	1✔
25	self.preferred_by = frozenset(self.preferred_by)	1✔
26	self.designated_for = frozenset(self.designated_for)	1✔
27
28	@cached_property	1✔
29	def required_skill(self):	1✔
30	graph = QualificationUniverse.get_graph()	1✔
31	required_qualification_uuids = [	1✔
32	qualification.uuid for qualification in self.required_qualifications
33	]
34	return set(graph.spread_from(required_qualification_uuids))	1✔
35
36	@cached_property	1✔
37	def skill_level(self):	1✔
38	return skill_level(self.required_skill)	1✔
39
40
41	def skill_level(qualifications):	1✔
42	"""
43	skill_level is some float in the range [0,1] that expresses
44	how far up in the qualification hierarchy the required qualifications are.
45	"""
46	if not qualifications:	1!
47	return 0	×
48	qualifications = list(qualifications)	1✔
49	# convert to uuids
50	if isinstance(qualifications[0], Qualification):	1✔
51	qualifications = [q.uuid for q in qualifications]	1✔
52	graph = QualificationUniverse.get_graph()	1✔
53	required_skill = set(graph.spread_from(qualifications))	1✔
54	# all_skill is qualifications up and down from the required ones
55	all_skill = required_skill \| set(graph.spread_reverse(qualifications))	1✔
56	return len(required_skill) / len(all_skill)	1✔
57
58
59	class Matching:	1✔
60
61	def __init__(self, participants, positions, pairings):	1✔
62	self.pairings = pairings	1✔
63	self.participants = participants	1✔
64	self.positions = positions	1✔
65	self.unpaired_participants = set(participants)	1✔
66	self.unpaired_positions = list(	1✔
67	sorted(positions, key=lambda position: (-position.skill_level, position.id))
68	)
69	for participant, position in pairings:	1✔
70	self.unpaired_participants.remove(participant)	1✔
71	self.unpaired_positions.remove(position)	1✔
72	self._participations = None	1✔
73
74	@property	1✔
75	def participations(self):	1✔
76	if self._participations is None:	1!
77	raise ValueError("participations not attached")	×
78	return self._participations	1✔
79
80	def attach_participations(self, participations):	1✔
81	self._participations = participations	1✔
82
83	@cached_property	1✔
84	def _participation_by_participant(self):	1✔
85	return {p.participant: p for p in self.participations}	1✔
86
87	@cached_property	1✔
88	def participation_pairings(self):	1✔
89	return [	1✔
90	(self._participation_by_participant[participant], position)
91	for participant, position in self.pairings
92	]
93
94	@cached_property	1✔
95	def unpaired_participations(self):	1✔
96	return [	1✔
97	participation
98	for participation in self.participations
99	if participation.participant in self.unpaired_participants
100	]
101
102	def participation_for_position(self, position_id):	1✔
NEW 103	for participation, position in self.participation_pairings:	×
NEW 104	if position.id == position_id:	×
NEW 105	return participation	×
106
107
108	def score_pairing(	1✔
109	participant: AbstractParticipant, position: Position, number_of_participants=1_000_000
110	):
111	"""
112	Score the pairing of participant and position.
113	Skill here means a set of qualification.
114	"""
115	# number of participants is used as a scoring value to make sure that in adverse cases
116	# a matching with a someone unqualified pairings gets a worse score than a matching with more valid pairings
117	# (similar for prefers and skill level). It can be arbitrarily big, but must be at least bigger than the sum of
118	# all the other constants used in the score.
119	padded_participant_count = 10 + 2 * number_of_participants	1✔
120	base_score = 1.0	1✔
121	preferred_value = 3.0	1✔
122	max_skill_value = 1.0	1✔
123	max_aux_value = 2.0	1✔
124	required_value = padded_participant_count * sum(	1✔
125	(base_score, preferred_value, max_skill_value, max_aux_value)
126	)
127	designated_value = required_value**2	1✔
128	unqualified_penalty = designated_value**2	1✔
129
130	is_designated = participant in position.designated_for	1✔
131
132	if not is_designated and not position.required_skill <= participant.skill:	1✔
133	# the participant does not have some required skill
134	return -unqualified_penalty # avoid matching unqualified participants	1✔
135
136	score = base_score	1✔
137	if is_designated:	1!
NEW 138	score += designated_value	×
139	if participant in position.preferred_by:	1✔
140	score += preferred_value	1✔
141	if position.required:	1✔
142	score += required_value	1✔
143	score += position.skill_level * max_skill_value	1✔
144	score += position.aux_score * max_aux_value	1✔
145	return score	1✔
146
147
148	def match_participants_to_positions(	1✔
149	participants: Collection[AbstractParticipant],
150	positions: Collection[Position],
151	) -> Matching:
152	participants = list(participants)	1✔
153	positions = list(positions)	1✔
154	costs = csr_matrix(	1✔
155	[
156	[
157	-score_pairing(participant, position, number_of_participants=len(participants))
158	for position in positions
159	]
160	for participant in participants
161	]
162	)
163	matching = min_weight_full_bipartite_matching(costs)	1✔
164	pairings = set()	1✔
165	for par_idx, pos_idx in zip(*matching):	1✔
166	cost = costs[par_idx, pos_idx]	1✔
167	if cost <= 0:	1✔
168	pairings.add((participants[par_idx], positions[pos_idx]))	1✔
169	return Matching(participants, positions, pairings)	1✔

ephios-dev / ephios / 10794283251

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous