Skip to content

Session models

session.models

Session

Bases: Model

A model defining a session of an experiment block of a participant

Attributes:

Name Type Description
block Block

each session is tied to a block

participant Participant

each session is tied to a participant

playlist Playlist

most sessions will also be tied to a playlist

started_at datetime

a timestamp when a session is created, auto-populated

finished_at datetime

a timestamp of when session.finish() was called

json_data json

a field to keep track of progress through a blocks’ rules in a session

final_score float

the final score of the session, usually the sum of all Result objects on the session

Source code in session/models.py
  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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
class Session(models.Model):
    """A model defining a session of an experiment block of a participant

    Attributes:
        block (experiment.models.Block): each session is tied to a block
        participant (participant.models.Participant): each session is tied to a participant
        playlist (section.models.Playlist): most sessions will also be tied to a playlist
        started_at (datetime): a timestamp when a session is created, auto-populated
        finished_at (datetime): a timestamp of when `session.finish()` was called
        json_data (json): a field to keep track of progress through a blocks' rules in a session
        final_score (float): the final score of the session, usually the sum of all `Result` objects on the session
    """
    block = models.ForeignKey("experiment.Block", on_delete=models.CASCADE, blank=True, null=True)
    participant = models.ForeignKey("participant.Participant", on_delete=models.CASCADE)
    playlist = models.ForeignKey("section.Playlist", on_delete=models.SET_NULL, blank=True, null=True)

    started_at = models.DateTimeField(db_index=True, default=timezone.now)
    finished_at = models.DateTimeField(db_index=True, default=None, null=True, blank=True)
    json_data = models.JSONField(default=dict, blank=True, null=True)
    final_score = models.FloatField(db_index=True, default=0.0)

    def __str__(self):
        return "Session {}".format(self.id)

    def result_count(self) -> int:
        """
        Returns:
            number of results for this session
        """
        return self.result_set.count()

    result_count.short_description = "Results"

    def block_rules(self):
        """
        Returns:
            (experiment.rules.Base) rules class to be used for this session
        """
        return self.block.get_rules()

    def finish(self):
        """Finish current session with the following steps:

        1. set the `finished_at` timestamp to the current moment

        2. set the `final_score` field to the sum of all results' scores
        """
        self.finished_at = timezone.now()
        self.final_score = self.total_score()

    def get_rounds_passed(self, counted_result_keys: list = []) -> int:
        """Get number of rounds passed, measured by the number of results on this session,
        taking into account the `counted_result_keys` array that may be defined per rules file

        Attributes:
            counted_result_keys: array of the Result.question_key strings which should be taken into account for counting rounds; if empty, all results will be counted.

        Returns:
            number of results, filtered by `counted_result_keys`, if supplied
        """
        results = self.result_set
        if counted_result_keys:
            results = results.filter(question_key__in=counted_result_keys)
        return results.count()

    def get_used_song_ids(self, exclude: dict = {}) -> Iterable[int]:
        """Get a list of song ids already used in this session

        Attributes:
            exclude: a dictionary by which to exclude specific results in this session, using [Django's querying syntax](https://docs.djangoproject.com/en/4.2/topics/db/queries/)

        Returns:
            a list of song ids from the sections of this session's results
        """
        return (res.section.song.id for res in self.result_set.exclude(**exclude).filter(section__isnull=False))

    def get_unused_song_ids(self, filter_by: dict = {}) -> Iterable[int]:
        """Get a list of unused song ids from this session's playlist

        Attributes:
            filter_by: a dictionary by which to select sections from the playlist (e.g., a certain tag), using [Django's querying syntax](https://docs.djangoproject.com/en/4.2/topics/db/queries/)

        Returns:
            a list of song ids which haven't been used in this session yet
        """
        # Get all song ids from the current playlist
        song_ids = (
            self.playlist.section_set.filter(**filter_by).order_by("song").values_list("song_id", flat=True).distinct()
        )
        # Get all song ids from results
        used_song_ids = self.get_used_song_ids()
        return list(set(song_ids) - set(used_song_ids))

    def last_result(self, question_keys: list[str] = []) -> Union[Result, None]:
        """
        Utility function to retrieve the last result, optionally filtering by relevant question keys.
        If more than one result needs to be processed, or for more advanced filtering,
        you can refer to the results on a session by `session.result_set` and query using the
        [Django's querying syntax](https://docs.djangoproject.com/en/4.2/topics/db/queries/)

        Attributes:
            question_keys: array of Result.question_key strings to specify whish results should be taken into account; if empty, return last result, irrespective of its question_key

        Returns:
            last relevant [Result](result_models.md#Result) object added to the database for this session
        """
        results = self.result_set
        if not results.count():
            return None
        if question_keys:
            results = results.filter(question_key__in=question_keys)
        return results.order_by("-created_at").first()

    def last_section(self, question_keys: list[str] = []) -> Union[Section, None]:
        """
        Utility function to retrieve the last section played in the session, optinally filtering by result question keys.
        Uses [last_result](session_models.md#Session.last_result) underneath.

        Attributes:
            question_keys: array of the Result.question_key strings whish should be taken into account; if empty, return last section, irrespective of question_key

        Returns:
            Section tied to previous result, if that result has a score and section, else None
        """
        result = self.last_result(question_keys)
        if result and result.section and result.score is not None:
            return result.section
        return None

    def last_score(self, question_keys: list[str] = []) -> float:
        """
        Utility function to retrieve last score logged to the session, optionally filtering by result question keys.
        Uses `last_result` underneath.

        Attributes:
            question_keys: array of the Result.question_key strings whish should be taken into account; if empty, return last score, irrespective of question_key

        Returns:
            score of last result, or return 0 if there are no results yet
        """
        result = self.last_result(question_keys)
        if result:
            return result.score
        return 0

    def last_song(self, question_keys: list[str] = []) -> str:
        """
        Utility function to retrieve label (artist - name) of last song played in session, optionally filtering by result question keys.
        Uses `last_result` underneath.

        Attributes:
            question_keys: array of the Result.question_key strings whish should be taken into account; if empty, return last played song, irrespective of question_key

        Returns:
            artist and name of section tied to previous result, if available, or an empty string
        """
        section = self.last_section(question_keys)
        if section:
            return section.song_label()
        return ""

    def percentile_rank(self, exclude_unfinished: bool) -> float:
        """
        Returns:
            Percentile rank of this session for the associated block, based on `final_score`
        """
        session_set = self.block.session_set
        if exclude_unfinished:
            session_set = session_set.filter(finished_at__isnull=False)
        n_session = session_set.count()
        if n_session == 0:
            return 0.0  # Should be impossible but avoids x/0
        n_lte = session_set.filter(final_score__lte=self.final_score).count()
        n_eq = session_set.filter(final_score=self.final_score).count()
        return 100.0 * (n_lte - (0.5 * n_eq)) / n_session

    def rank(self) -> int:
        """
        Returns:
            rank of the current session for the associated block, based on `final_score`
        """
        return (
            self.block.session_set.filter(final_score__gte=self.final_score)
            .values("final_score")
            .annotate(total=models.Count("final_score"))
            .count()
        )

    def rounds_complete(self, counted_result_keys: list[str] = []) -> bool:
        """
        Attributes:
            counted_result_keys: array of the Result.question_key strings which should be taken into account for counting rounds; if empty, all results will be counted.

        Returns:
            True if there are results for each experiment round
        """
        return self.get_rounds_passed(counted_result_keys) >= self.block.rounds

    def total_score(self) -> float:
        """
        Returns:
            sum of all result scores
        """
        score = self.result_set.aggregate(models.Sum("score"))
        return self.block.bonus_points + (
            score["score__sum"] if score["score__sum"] else 0
        )

    def save_json_data(self, data: dict):
        """Merge data with json_data, overwriting duplicate keys.

        Attributes:
            data: a dictionary of data to save to the `json_data` field
        """
        self.json_data.update(data)
        self.save()

    def _export_admin(self):
        """Export data for admin"""
        return {
            "session_id": self.id,
            "participant": self.participant.id,
            "started_at": self.started_at.isoformat(),
            "finished_at": self.finished_at.isoformat() if self.finished_at else None,
            "json_data": self.json_data,
            "results": [result._export_admin() for result in self.result_set.all()],
        }

    def _is_finished(self) -> bool:
        """
        Returns:
            a boolean to indicate whether the session is finished
        """
        return self.finished_at

block_rules()

Returns:

Type Description

(experiment.rules.Base) rules class to be used for this session

Source code in session/models.py
42
43
44
45
46
47
def block_rules(self):
    """
    Returns:
        (experiment.rules.Base) rules class to be used for this session
    """
    return self.block.get_rules()

finish()

Finish current session with the following steps:

  1. set the finished_at timestamp to the current moment

  2. set the final_score field to the sum of all results’ scores

Source code in session/models.py
49
50
51
52
53
54
55
56
57
def finish(self):
    """Finish current session with the following steps:

    1. set the `finished_at` timestamp to the current moment

    2. set the `final_score` field to the sum of all results' scores
    """
    self.finished_at = timezone.now()
    self.final_score = self.total_score()

get_rounds_passed(counted_result_keys=[])

Get number of rounds passed, measured by the number of results on this session, taking into account the counted_result_keys array that may be defined per rules file

Attributes:

Name Type Description
counted_result_keys

array of the Result.question_key strings which should be taken into account for counting rounds; if empty, all results will be counted.

Returns:

Type Description
int

number of results, filtered by counted_result_keys, if supplied

Source code in session/models.py
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_rounds_passed(self, counted_result_keys: list = []) -> int:
    """Get number of rounds passed, measured by the number of results on this session,
    taking into account the `counted_result_keys` array that may be defined per rules file

    Attributes:
        counted_result_keys: array of the Result.question_key strings which should be taken into account for counting rounds; if empty, all results will be counted.

    Returns:
        number of results, filtered by `counted_result_keys`, if supplied
    """
    results = self.result_set
    if counted_result_keys:
        results = results.filter(question_key__in=counted_result_keys)
    return results.count()

get_unused_song_ids(filter_by={})

Get a list of unused song ids from this session’s playlist

Attributes:

Name Type Description
filter_by

a dictionary by which to select sections from the playlist (e.g., a certain tag), using Django’s querying syntax

Returns:

Type Description
Iterable[int]

a list of song ids which haven’t been used in this session yet

Source code in session/models.py
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def get_unused_song_ids(self, filter_by: dict = {}) -> Iterable[int]:
    """Get a list of unused song ids from this session's playlist

    Attributes:
        filter_by: a dictionary by which to select sections from the playlist (e.g., a certain tag), using [Django's querying syntax](https://docs.djangoproject.com/en/4.2/topics/db/queries/)

    Returns:
        a list of song ids which haven't been used in this session yet
    """
    # Get all song ids from the current playlist
    song_ids = (
        self.playlist.section_set.filter(**filter_by).order_by("song").values_list("song_id", flat=True).distinct()
    )
    # Get all song ids from results
    used_song_ids = self.get_used_song_ids()
    return list(set(song_ids) - set(used_song_ids))

get_used_song_ids(exclude={})

Get a list of song ids already used in this session

Attributes:

Name Type Description
exclude

a dictionary by which to exclude specific results in this session, using Django’s querying syntax

Returns:

Type Description
Iterable[int]

a list of song ids from the sections of this session’s results

Source code in session/models.py
74
75
76
77
78
79
80
81
82
83
def get_used_song_ids(self, exclude: dict = {}) -> Iterable[int]:
    """Get a list of song ids already used in this session

    Attributes:
        exclude: a dictionary by which to exclude specific results in this session, using [Django's querying syntax](https://docs.djangoproject.com/en/4.2/topics/db/queries/)

    Returns:
        a list of song ids from the sections of this session's results
    """
    return (res.section.song.id for res in self.result_set.exclude(**exclude).filter(section__isnull=False))

last_result(question_keys=[])

Utility function to retrieve the last result, optionally filtering by relevant question keys. If more than one result needs to be processed, or for more advanced filtering, you can refer to the results on a session by session.result_set and query using the Django’s querying syntax

Attributes:

Name Type Description
question_keys

array of Result.question_key strings to specify whish results should be taken into account; if empty, return last result, irrespective of its question_key

Returns:

Type Description
Union[Result, None]

last relevant Result object added to the database for this session

Source code in session/models.py
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def last_result(self, question_keys: list[str] = []) -> Union[Result, None]:
    """
    Utility function to retrieve the last result, optionally filtering by relevant question keys.
    If more than one result needs to be processed, or for more advanced filtering,
    you can refer to the results on a session by `session.result_set` and query using the
    [Django's querying syntax](https://docs.djangoproject.com/en/4.2/topics/db/queries/)

    Attributes:
        question_keys: array of Result.question_key strings to specify whish results should be taken into account; if empty, return last result, irrespective of its question_key

    Returns:
        last relevant [Result](result_models.md#Result) object added to the database for this session
    """
    results = self.result_set
    if not results.count():
        return None
    if question_keys:
        results = results.filter(question_key__in=question_keys)
    return results.order_by("-created_at").first()

last_score(question_keys=[])

Utility function to retrieve last score logged to the session, optionally filtering by result question keys. Uses last_result underneath.

Attributes:

Name Type Description
question_keys

array of the Result.question_key strings whish should be taken into account; if empty, return last score, irrespective of question_key

Returns:

Type Description
float

score of last result, or return 0 if there are no results yet

Source code in session/models.py
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def last_score(self, question_keys: list[str] = []) -> float:
    """
    Utility function to retrieve last score logged to the session, optionally filtering by result question keys.
    Uses `last_result` underneath.

    Attributes:
        question_keys: array of the Result.question_key strings whish should be taken into account; if empty, return last score, irrespective of question_key

    Returns:
        score of last result, or return 0 if there are no results yet
    """
    result = self.last_result(question_keys)
    if result:
        return result.score
    return 0

last_section(question_keys=[])

Utility function to retrieve the last section played in the session, optinally filtering by result question keys. Uses last_result underneath.

Attributes:

Name Type Description
question_keys

array of the Result.question_key strings whish should be taken into account; if empty, return last section, irrespective of question_key

Returns:

Type Description
Union[Section, None]

Section tied to previous result, if that result has a score and section, else None

Source code in session/models.py
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def last_section(self, question_keys: list[str] = []) -> Union[Section, None]:
    """
    Utility function to retrieve the last section played in the session, optinally filtering by result question keys.
    Uses [last_result](session_models.md#Session.last_result) underneath.

    Attributes:
        question_keys: array of the Result.question_key strings whish should be taken into account; if empty, return last section, irrespective of question_key

    Returns:
        Section tied to previous result, if that result has a score and section, else None
    """
    result = self.last_result(question_keys)
    if result and result.section and result.score is not None:
        return result.section
    return None

last_song(question_keys=[])

Utility function to retrieve label (artist - name) of last song played in session, optionally filtering by result question keys. Uses last_result underneath.

Attributes:

Name Type Description
question_keys

array of the Result.question_key strings whish should be taken into account; if empty, return last played song, irrespective of question_key

Returns:

Type Description
str

artist and name of section tied to previous result, if available, or an empty string

Source code in session/models.py
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def last_song(self, question_keys: list[str] = []) -> str:
    """
    Utility function to retrieve label (artist - name) of last song played in session, optionally filtering by result question keys.
    Uses `last_result` underneath.

    Attributes:
        question_keys: array of the Result.question_key strings whish should be taken into account; if empty, return last played song, irrespective of question_key

    Returns:
        artist and name of section tied to previous result, if available, or an empty string
    """
    section = self.last_section(question_keys)
    if section:
        return section.song_label()
    return ""

percentile_rank(exclude_unfinished)

Returns:

Type Description
float

Percentile rank of this session for the associated block, based on final_score

Source code in session/models.py
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def percentile_rank(self, exclude_unfinished: bool) -> float:
    """
    Returns:
        Percentile rank of this session for the associated block, based on `final_score`
    """
    session_set = self.block.session_set
    if exclude_unfinished:
        session_set = session_set.filter(finished_at__isnull=False)
    n_session = session_set.count()
    if n_session == 0:
        return 0.0  # Should be impossible but avoids x/0
    n_lte = session_set.filter(final_score__lte=self.final_score).count()
    n_eq = session_set.filter(final_score=self.final_score).count()
    return 100.0 * (n_lte - (0.5 * n_eq)) / n_session

rank()

Returns:

Type Description
int

rank of the current session for the associated block, based on final_score

Source code in session/models.py
185
186
187
188
189
190
191
192
193
194
195
def rank(self) -> int:
    """
    Returns:
        rank of the current session for the associated block, based on `final_score`
    """
    return (
        self.block.session_set.filter(final_score__gte=self.final_score)
        .values("final_score")
        .annotate(total=models.Count("final_score"))
        .count()
    )

result_count()

Returns:

Type Description
int

number of results for this session

Source code in session/models.py
33
34
35
36
37
38
def result_count(self) -> int:
    """
    Returns:
        number of results for this session
    """
    return self.result_set.count()

rounds_complete(counted_result_keys=[])

Attributes:

Name Type Description
counted_result_keys

array of the Result.question_key strings which should be taken into account for counting rounds; if empty, all results will be counted.

Returns:

Type Description
bool

True if there are results for each experiment round

Source code in session/models.py
197
198
199
200
201
202
203
204
205
def rounds_complete(self, counted_result_keys: list[str] = []) -> bool:
    """
    Attributes:
        counted_result_keys: array of the Result.question_key strings which should be taken into account for counting rounds; if empty, all results will be counted.

    Returns:
        True if there are results for each experiment round
    """
    return self.get_rounds_passed(counted_result_keys) >= self.block.rounds

save_json_data(data)

Merge data with json_data, overwriting duplicate keys.

Attributes:

Name Type Description
data

a dictionary of data to save to the json_data field

Source code in session/models.py
217
218
219
220
221
222
223
224
def save_json_data(self, data: dict):
    """Merge data with json_data, overwriting duplicate keys.

    Attributes:
        data: a dictionary of data to save to the `json_data` field
    """
    self.json_data.update(data)
    self.save()

total_score()

Returns:

Type Description
float

sum of all result scores

Source code in session/models.py
207
208
209
210
211
212
213
214
215
def total_score(self) -> float:
    """
    Returns:
        sum of all result scores
    """
    score = self.result_set.aggregate(models.Sum("score"))
    return self.block.bonus_points + (
        score["score__sum"] if score["score__sum"] else 0
    )