telemetry.core.logic

Docstring for telemetry.core.logic

This module provides the logic for aggregating and analysing the data parsed by the parsing module, namely sorting by difficulty, getting averages by difficulty, etc...

This functionality is provided by the EventLogicEngine class.

  1"""
  2Docstring for telemetry.core.logic
  3
  4This module provides the logic for aggregating and analysing the data
  5parsed by the parsing module, namely sorting by difficulty, getting 
  6averages by difficulty, etc...
  7
  8This functionality is provided by the EventLogicEngine class.
  9"""
 10
 11from pathlib import Path
 12
 13from telemetry.core.events import (
 14    BossEncounterComplete,
 15    BossEncounterFail,
 16    BossEncounterStart,
 17    BuyUpgrade,
 18    Difficulty,
 19    EndSession,
 20    GainCoin,
 21    KillEnemy,
 22    NormalEncounterComplete,
 23    NormalEncounterFail,
 24    NormalEncounterStart,
 25    SettingsChange,
 26    StartSession,
 27)
 28from telemetry.core.parsing import parse_file, ValidEvent
 29
 30
 31class EventLogicEngine:
 32    def __init__(self):
 33        self.start_session_events: set[
 34            StartSession
 35        ] = set()
 36        self.end_session_events: set[
 37            EndSession
 38        ] = set()
 39        self.normal_encounter_start_events: set[
 40            NormalEncounterStart
 41        ] = set()
 42        self.normal_encounter_complete_events: set[
 43            NormalEncounterComplete
 44        ] = set()
 45        self.normal_encounter_fail_events: set[
 46            NormalEncounterFail
 47        ] = set()
 48        self.boss_encounter_start_events: set[
 49            BossEncounterStart
 50        ] = set()
 51        self.boss_encounter_complete_events: set[
 52            BossEncounterComplete
 53        ] = set()
 54        self.boss_encounter_fail_events: set[
 55            BossEncounterFail
 56        ] = set()
 57        self.gain_coin_events: set[
 58            GainCoin
 59        ] = set()
 60        self.buy_upgrade_events: set[
 61            BuyUpgrade
 62        ] = set()
 63        self.settings_change_events: set[
 64            SettingsChange
 65        ] = set()
 66        self.kill_enemy_events: set[
 67            KillEnemy
 68        ] = set()
 69
 70        self._attributes = [ 
 71            self.start_session_events, 
 72            self.end_session_events,
 73            self.normal_encounter_start_events,
 74            self.normal_encounter_complete_events,
 75            self.normal_encounter_fail_events,
 76            self.boss_encounter_start_events,
 77            self.boss_encounter_complete_events,
 78            self.boss_encounter_fail_events,
 79            self.gain_coin_events,
 80            self.buy_upgrade_events,
 81            self.settings_change_events,
 82            self.kill_enemy_events
 83        ]
 84
 85
 86    def categorise_events(self, filename: Path) -> None:
 87        """
 88        Creates objects from the json file provided. 
 89
 90        :param filename: json file with custom schema.
 91        :type filename: Path
 92        """
 93        # TODO: We would need to implement __eq__ methods for our 
 94        # event classes if we want the set() functionality of 
 95        # avoiding duplicates. Until then, temporary fix: clear the 
 96        # attributes before reading them back in again.
 97        for attr in self._attributes:
 98            attr.clear()
 99        
100        events: list[ValidEvent] = parse_file(filename)
101        for event in events:
102            if isinstance(event, StartSession):
103                self.start_session_events.add(event)
104            elif isinstance(event, EndSession):
105                self.end_session_events.add(event)
106            elif isinstance(event, NormalEncounterStart):
107                self.normal_encounter_start_events.add(event)
108            elif isinstance(event, NormalEncounterComplete):
109                self.normal_encounter_complete_events.add(event)
110            elif isinstance(event, NormalEncounterFail):
111                self.normal_encounter_fail_events.add(event)
112            elif isinstance(event, BossEncounterStart):
113                self.boss_encounter_start_events.add(event)
114            elif isinstance(event, BossEncounterComplete):
115                self.boss_encounter_complete_events.add(event)
116            elif isinstance(event, BossEncounterFail):
117                self.boss_encounter_fail_events.add(event)
118            elif isinstance(event, GainCoin):
119                self.gain_coin_events.add(event)
120            elif isinstance(event, BuyUpgrade):
121                self.buy_upgrade_events.add(event)
122            elif isinstance(event, SettingsChange):
123                self.settings_change_events.add(event)
124            elif isinstance(event, KillEnemy):
125                self.kill_enemy_events.add(event)
126
127
128    def fail_difficulty_spikes(self) -> dict[int, int]:
129        """
130        Output failure rate by stage.
131        
132        :return: dictionary of key stage number and value number of 
133        failures.
134        :rtype: dict[int, int]
135        """
136        difficulty_output = {stage_number: 0 for stage_number in range(1,11)}
137        for event in self.normal_encounter_fail_events:
138            difficulty_output[event.stage_number] += 1
139        for event in self.boss_encounter_fail_events:
140            difficulty_output[event.stage_number] += 1
141        return difficulty_output
142    
143
144    def get_number_of_session_starts(self) -> int:
145        """
146        Get the number of start session events.        
147
148        :return: Returns the number of unique session start events.
149        :rtype: int
150        """
151        return len(self.start_session_events)
152    
153    
154    def get_unique_userIDs(self) -> set[int]:
155        """
156        Returns the set of unique user IDs.
157        
158        :return: set of unique user IDs.
159        :rtype: set[int]
160        """
161        uniqueIDs = set()
162        for event in self.start_session_events:
163            uniqueIDs.add(event.userID)
164        return uniqueIDs
165    
166
167    def count_starts(self, stage_number) -> int:
168        """
169        Counts the number of starts of a given stage.
170        
171        :param stage_number: stage number in question.
172        :return: number of starts of that stage.
173        :rtype: int
174        """
175        start_count = 0
176        if stage_number in [3, 6, 9, 10]:
177            for start_event in self.boss_encounter_start_events:
178                start_count += start_event.stage_number == stage_number
179            return start_count
180        for start_event in self.normal_encounter_start_events:
181            start_count += start_event.stage_number == stage_number
182        return start_count     
183       
184
185    def count_fails(self, stage_number) -> int:
186        """
187        Counts the number of failures on a given stage.
188        
189        :param stage_number: stage number in question.
190        :return: number of fails at that stage.
191        :rtype: int
192        """
193        fail_count = 0
194        if stage_number in [3, 6, 9, 10]:
195            for fail_event in self.boss_encounter_fail_events:
196                fail_count += fail_event.stage_number == stage_number
197            return fail_count
198        for fail_event in self.normal_encounter_fail_events:
199            fail_count += fail_event.stage_number == stage_number
200        return fail_count
201    
202    
203    def funnel_view(self) -> dict[int, int]:
204        """
205        Output number of players passing a given stage.
206        
207        :return: dictionary of key stage number and value number of 
208        players left.
209        :rtype: dict[int, int]
210        """
211        return {
212            stage_number: 
213            self.count_starts(stage_number) - self.count_fails(stage_number) 
214            for stage_number in range(1,11)
215        }
216    
217    
218    def health_per_stage(self, sessionID: int) -> dict[int, int]:
219        """
220        Output the health a session has per stage.
221        
222        :param sessionID: session stage to check.
223        :type sessionID: int
224        :return: dictionary of key stage number and value player HP
225        remaining.
226        :rtype: dict[int, int]
227        """
228        health_remaining_per_stage = {stage_number: 0 
229                                      for stage_number in range(1,11)}
230        for event in self.normal_encounter_complete_events:
231            if event.sessionID == sessionID:
232                health_remaining_per_stage[
233                    event.stage_number
234                ] = event.player_HP_remaining
235        for event in self.boss_encounter_complete_events:
236            if event.sessionID == sessionID:
237                health_remaining_per_stage[
238                    event.stage_number
239                ] = event.player_HP_remaining
240        return health_remaining_per_stage
241    
242    
243    def get_difficulty(self, sessionID: int) -> Difficulty:
244        """
245        Get the difficulty for a given session.
246        
247        :param sessionID: sessionID of the session to get difficulty of.
248        :type sessionID: int
249        :return: Difficulty value of the session.
250        :rtype: Difficulty
251        """
252        for start_event in self.start_session_events:
253            if start_event.sessionID == sessionID:
254                return start_event.difficulty
255        raise RuntimeError("No session start event for provided " \
256        f"session ID: {sessionID}")
257    
258    
259    def get_sessionIDs_of_difficulty(self, difficulty: Difficulty) -> list[int]:
260        """
261        Get the list of all sessionIDs of a given difficulty.
262        
263        :param difficulty: The difficulty level to search for.
264        :type difficulty: Difficulty
265        :return: The list of sessionIDs with the given difficulty.
266        :rtype: list[int]
267        """
268        difficulty_sessionIDs: list[int] = []
269        for start_event in self.start_session_events:
270            if start_event.difficulty == difficulty:
271                difficulty_sessionIDs.append(start_event.sessionID)
272        return difficulty_sessionIDs
273        
274
275    def get_health_per_stage_by_difficulty(
276            self,
277            difficulty: Difficulty
278    ) -> list[dict[int, int]]:
279        """
280        Returns a health_per_stage dictionary for all sessions with a
281        given difficulty.
282        
283        :param difficulty: Given difficulty.
284        :type difficulty: Difficulty
285        :return: List of health per stage for each session with the
286        specified difficulty level.
287        :rtype: list[dict[int, int]]
288        """
289        health_per_stage = []
290        for sessionID in self.get_sessionIDs_of_difficulty(difficulty):
291            health_per_stage.append(self.health_per_stage(sessionID))
292        return health_per_stage
293    
294
295    def compare_health_per_stage_per_difficulty(
296            self
297    ) -> dict[Difficulty, list[dict[int, int]]]:
298        """
299        Get a dictionary with difficulty level as the key, and the list 
300        of the health_per_stage objects of all sessions with that 
301        difficulty level as the value.
302        
303        :return: Dictionary of difficulty level to list of all 
304        health_per_stage dictionaries of all sessions with the given
305        difficulty level.
306        :rtype: dict[Difficulty, list[dict[int, int]]]
307        """
308        return {diff: self.get_health_per_stage_by_difficulty(diff) 
309                for diff in Difficulty}
310    
311
312    def get_coins_gained_per_stage(
313            self, 
314            sessionID: int
315    ) -> dict[int, int]:
316        """
317        Get the number of coins gained at each stage for a given 
318        session.
319        
320        :param sessionID: sessionID in question.
321        :type sessionID: int
322        :return: Stage number -> coins gained at that stage.
323        :rtype: dict[int, int]
324        """
325        coins_gained_per_stage: dict[int, int] = {i: 0 for i in range(1, 11)}
326        for event in self.gain_coin_events:
327            if event.sessionID == sessionID:
328                coins_gained_per_stage[event.stage_number] += event.coins_gained
329        return coins_gained_per_stage
330    
331
332    def get_coins_per_stage_by_difficulty(
333            self,
334            difficulty: Difficulty
335    ) -> list[dict[int, int]]:
336        """
337        Returns the coins gained per stage dictionary for all sessions 
338        with a given difficulty.
339        
340        :param difficulty: Given difficulty.
341        :type difficulty: Difficulty
342        :return: List of coins per stage for each session with the
343        specified difficulty level.
344        :rtype: list[dict[int, int]]
345        """
346        coins_per_stage = []
347        for sessionID in self.get_sessionIDs_of_difficulty(difficulty):
348            coins_per_stage.append(self.get_coins_gained_per_stage(sessionID))
349        return coins_per_stage
350    
351
352    def compare_coins_per_stage_per_difficulty(
353            self
354    ) -> dict[Difficulty, list[dict[int, int]]]:
355        """
356        Get a dictionary with difficulty level as the key, and the list 
357        of the coins gained per stage dictionary objects of all sessions
358        with that difficulty level as the value.
359        
360        :return: Dictionary of difficulty level to list of all 
361        coins gained per stage dictionaries of all sessions with the 
362        given difficulty level.
363        :rtype: dict[Difficulty, list[dict[int, int]]]
364        """
365        return {diff: self.get_coins_per_stage_by_difficulty(diff) 
366                for diff in Difficulty}
class EventLogicEngine: 
 32class EventLogicEngine:
 33    def __init__(self):
 34        self.start_session_events: set[
 35            StartSession
 36        ] = set()
 37        self.end_session_events: set[
 38            EndSession
 39        ] = set()
 40        self.normal_encounter_start_events: set[
 41            NormalEncounterStart
 42        ] = set()
 43        self.normal_encounter_complete_events: set[
 44            NormalEncounterComplete
 45        ] = set()
 46        self.normal_encounter_fail_events: set[
 47            NormalEncounterFail
 48        ] = set()
 49        self.boss_encounter_start_events: set[
 50            BossEncounterStart
 51        ] = set()
 52        self.boss_encounter_complete_events: set[
 53            BossEncounterComplete
 54        ] = set()
 55        self.boss_encounter_fail_events: set[
 56            BossEncounterFail
 57        ] = set()
 58        self.gain_coin_events: set[
 59            GainCoin
 60        ] = set()
 61        self.buy_upgrade_events: set[
 62            BuyUpgrade
 63        ] = set()
 64        self.settings_change_events: set[
 65            SettingsChange
 66        ] = set()
 67        self.kill_enemy_events: set[
 68            KillEnemy
 69        ] = set()
 70
 71        self._attributes = [ 
 72            self.start_session_events, 
 73            self.end_session_events,
 74            self.normal_encounter_start_events,
 75            self.normal_encounter_complete_events,
 76            self.normal_encounter_fail_events,
 77            self.boss_encounter_start_events,
 78            self.boss_encounter_complete_events,
 79            self.boss_encounter_fail_events,
 80            self.gain_coin_events,
 81            self.buy_upgrade_events,
 82            self.settings_change_events,
 83            self.kill_enemy_events
 84        ]
 85
 86
 87    def categorise_events(self, filename: Path) -> None:
 88        """
 89        Creates objects from the json file provided. 
 90
 91        :param filename: json file with custom schema.
 92        :type filename: Path
 93        """
 94        # TODO: We would need to implement __eq__ methods for our 
 95        # event classes if we want the set() functionality of 
 96        # avoiding duplicates. Until then, temporary fix: clear the 
 97        # attributes before reading them back in again.
 98        for attr in self._attributes:
 99            attr.clear()
100        
101        events: list[ValidEvent] = parse_file(filename)
102        for event in events:
103            if isinstance(event, StartSession):
104                self.start_session_events.add(event)
105            elif isinstance(event, EndSession):
106                self.end_session_events.add(event)
107            elif isinstance(event, NormalEncounterStart):
108                self.normal_encounter_start_events.add(event)
109            elif isinstance(event, NormalEncounterComplete):
110                self.normal_encounter_complete_events.add(event)
111            elif isinstance(event, NormalEncounterFail):
112                self.normal_encounter_fail_events.add(event)
113            elif isinstance(event, BossEncounterStart):
114                self.boss_encounter_start_events.add(event)
115            elif isinstance(event, BossEncounterComplete):
116                self.boss_encounter_complete_events.add(event)
117            elif isinstance(event, BossEncounterFail):
118                self.boss_encounter_fail_events.add(event)
119            elif isinstance(event, GainCoin):
120                self.gain_coin_events.add(event)
121            elif isinstance(event, BuyUpgrade):
122                self.buy_upgrade_events.add(event)
123            elif isinstance(event, SettingsChange):
124                self.settings_change_events.add(event)
125            elif isinstance(event, KillEnemy):
126                self.kill_enemy_events.add(event)
127
128
129    def fail_difficulty_spikes(self) -> dict[int, int]:
130        """
131        Output failure rate by stage.
132        
133        :return: dictionary of key stage number and value number of 
134        failures.
135        :rtype: dict[int, int]
136        """
137        difficulty_output = {stage_number: 0 for stage_number in range(1,11)}
138        for event in self.normal_encounter_fail_events:
139            difficulty_output[event.stage_number] += 1
140        for event in self.boss_encounter_fail_events:
141            difficulty_output[event.stage_number] += 1
142        return difficulty_output
143    
144
145    def get_number_of_session_starts(self) -> int:
146        """
147        Get the number of start session events.        
148
149        :return: Returns the number of unique session start events.
150        :rtype: int
151        """
152        return len(self.start_session_events)
153    
154    
155    def get_unique_userIDs(self) -> set[int]:
156        """
157        Returns the set of unique user IDs.
158        
159        :return: set of unique user IDs.
160        :rtype: set[int]
161        """
162        uniqueIDs = set()
163        for event in self.start_session_events:
164            uniqueIDs.add(event.userID)
165        return uniqueIDs
166    
167
168    def count_starts(self, stage_number) -> int:
169        """
170        Counts the number of starts of a given stage.
171        
172        :param stage_number: stage number in question.
173        :return: number of starts of that stage.
174        :rtype: int
175        """
176        start_count = 0
177        if stage_number in [3, 6, 9, 10]:
178            for start_event in self.boss_encounter_start_events:
179                start_count += start_event.stage_number == stage_number
180            return start_count
181        for start_event in self.normal_encounter_start_events:
182            start_count += start_event.stage_number == stage_number
183        return start_count     
184       
185
186    def count_fails(self, stage_number) -> int:
187        """
188        Counts the number of failures on a given stage.
189        
190        :param stage_number: stage number in question.
191        :return: number of fails at that stage.
192        :rtype: int
193        """
194        fail_count = 0
195        if stage_number in [3, 6, 9, 10]:
196            for fail_event in self.boss_encounter_fail_events:
197                fail_count += fail_event.stage_number == stage_number
198            return fail_count
199        for fail_event in self.normal_encounter_fail_events:
200            fail_count += fail_event.stage_number == stage_number
201        return fail_count
202    
203    
204    def funnel_view(self) -> dict[int, int]:
205        """
206        Output number of players passing a given stage.
207        
208        :return: dictionary of key stage number and value number of 
209        players left.
210        :rtype: dict[int, int]
211        """
212        return {
213            stage_number: 
214            self.count_starts(stage_number) - self.count_fails(stage_number) 
215            for stage_number in range(1,11)
216        }
217    
218    
219    def health_per_stage(self, sessionID: int) -> dict[int, int]:
220        """
221        Output the health a session has per stage.
222        
223        :param sessionID: session stage to check.
224        :type sessionID: int
225        :return: dictionary of key stage number and value player HP
226        remaining.
227        :rtype: dict[int, int]
228        """
229        health_remaining_per_stage = {stage_number: 0 
230                                      for stage_number in range(1,11)}
231        for event in self.normal_encounter_complete_events:
232            if event.sessionID == sessionID:
233                health_remaining_per_stage[
234                    event.stage_number
235                ] = event.player_HP_remaining
236        for event in self.boss_encounter_complete_events:
237            if event.sessionID == sessionID:
238                health_remaining_per_stage[
239                    event.stage_number
240                ] = event.player_HP_remaining
241        return health_remaining_per_stage
242    
243    
244    def get_difficulty(self, sessionID: int) -> Difficulty:
245        """
246        Get the difficulty for a given session.
247        
248        :param sessionID: sessionID of the session to get difficulty of.
249        :type sessionID: int
250        :return: Difficulty value of the session.
251        :rtype: Difficulty
252        """
253        for start_event in self.start_session_events:
254            if start_event.sessionID == sessionID:
255                return start_event.difficulty
256        raise RuntimeError("No session start event for provided " \
257        f"session ID: {sessionID}")
258    
259    
260    def get_sessionIDs_of_difficulty(self, difficulty: Difficulty) -> list[int]:
261        """
262        Get the list of all sessionIDs of a given difficulty.
263        
264        :param difficulty: The difficulty level to search for.
265        :type difficulty: Difficulty
266        :return: The list of sessionIDs with the given difficulty.
267        :rtype: list[int]
268        """
269        difficulty_sessionIDs: list[int] = []
270        for start_event in self.start_session_events:
271            if start_event.difficulty == difficulty:
272                difficulty_sessionIDs.append(start_event.sessionID)
273        return difficulty_sessionIDs
274        
275
276    def get_health_per_stage_by_difficulty(
277            self,
278            difficulty: Difficulty
279    ) -> list[dict[int, int]]:
280        """
281        Returns a health_per_stage dictionary for all sessions with a
282        given difficulty.
283        
284        :param difficulty: Given difficulty.
285        :type difficulty: Difficulty
286        :return: List of health per stage for each session with the
287        specified difficulty level.
288        :rtype: list[dict[int, int]]
289        """
290        health_per_stage = []
291        for sessionID in self.get_sessionIDs_of_difficulty(difficulty):
292            health_per_stage.append(self.health_per_stage(sessionID))
293        return health_per_stage
294    
295
296    def compare_health_per_stage_per_difficulty(
297            self
298    ) -> dict[Difficulty, list[dict[int, int]]]:
299        """
300        Get a dictionary with difficulty level as the key, and the list 
301        of the health_per_stage objects of all sessions with that 
302        difficulty level as the value.
303        
304        :return: Dictionary of difficulty level to list of all 
305        health_per_stage dictionaries of all sessions with the given
306        difficulty level.
307        :rtype: dict[Difficulty, list[dict[int, int]]]
308        """
309        return {diff: self.get_health_per_stage_by_difficulty(diff) 
310                for diff in Difficulty}
311    
312
313    def get_coins_gained_per_stage(
314            self, 
315            sessionID: int
316    ) -> dict[int, int]:
317        """
318        Get the number of coins gained at each stage for a given 
319        session.
320        
321        :param sessionID: sessionID in question.
322        :type sessionID: int
323        :return: Stage number -> coins gained at that stage.
324        :rtype: dict[int, int]
325        """
326        coins_gained_per_stage: dict[int, int] = {i: 0 for i in range(1, 11)}
327        for event in self.gain_coin_events:
328            if event.sessionID == sessionID:
329                coins_gained_per_stage[event.stage_number] += event.coins_gained
330        return coins_gained_per_stage
331    
332
333    def get_coins_per_stage_by_difficulty(
334            self,
335            difficulty: Difficulty
336    ) -> list[dict[int, int]]:
337        """
338        Returns the coins gained per stage dictionary for all sessions 
339        with a given difficulty.
340        
341        :param difficulty: Given difficulty.
342        :type difficulty: Difficulty
343        :return: List of coins per stage for each session with the
344        specified difficulty level.
345        :rtype: list[dict[int, int]]
346        """
347        coins_per_stage = []
348        for sessionID in self.get_sessionIDs_of_difficulty(difficulty):
349            coins_per_stage.append(self.get_coins_gained_per_stage(sessionID))
350        return coins_per_stage
351    
352
353    def compare_coins_per_stage_per_difficulty(
354            self
355    ) -> dict[Difficulty, list[dict[int, int]]]:
356        """
357        Get a dictionary with difficulty level as the key, and the list 
358        of the coins gained per stage dictionary objects of all sessions
359        with that difficulty level as the value.
360        
361        :return: Dictionary of difficulty level to list of all 
362        coins gained per stage dictionaries of all sessions with the 
363        given difficulty level.
364        :rtype: dict[Difficulty, list[dict[int, int]]]
365        """
366        return {diff: self.get_coins_per_stage_by_difficulty(diff) 
367                for diff in Difficulty}
start_session_events: set[telemetry.core.events.StartSession]
end_session_events: set[telemetry.core.events.EndSession]
normal_encounter_start_events: set[telemetry.core.events.NormalEncounterStart]
normal_encounter_complete_events: set[telemetry.core.events.NormalEncounterComplete]
normal_encounter_fail_events: set[telemetry.core.events.NormalEncounterFail]
boss_encounter_start_events: set[telemetry.core.events.BossEncounterStart]
boss_encounter_complete_events: set[telemetry.core.events.BossEncounterComplete]
boss_encounter_fail_events: set[telemetry.core.events.BossEncounterFail]
gain_coin_events: set[telemetry.core.events.GainCoin]
buy_upgrade_events: set[telemetry.core.events.BuyUpgrade]
settings_change_events: set[telemetry.core.events.SettingsChange]
kill_enemy_events: set[telemetry.core.events.KillEnemy]
def categorise_events(self, filename: pathlib.Path) -> None: 
 87    def categorise_events(self, filename: Path) -> None:
 88        """
 89        Creates objects from the json file provided. 
 90
 91        :param filename: json file with custom schema.
 92        :type filename: Path
 93        """
 94        # TODO: We would need to implement __eq__ methods for our 
 95        # event classes if we want the set() functionality of 
 96        # avoiding duplicates. Until then, temporary fix: clear the 
 97        # attributes before reading them back in again.
 98        for attr in self._attributes:
 99            attr.clear()
100        
101        events: list[ValidEvent] = parse_file(filename)
102        for event in events:
103            if isinstance(event, StartSession):
104                self.start_session_events.add(event)
105            elif isinstance(event, EndSession):
106                self.end_session_events.add(event)
107            elif isinstance(event, NormalEncounterStart):
108                self.normal_encounter_start_events.add(event)
109            elif isinstance(event, NormalEncounterComplete):
110                self.normal_encounter_complete_events.add(event)
111            elif isinstance(event, NormalEncounterFail):
112                self.normal_encounter_fail_events.add(event)
113            elif isinstance(event, BossEncounterStart):
114                self.boss_encounter_start_events.add(event)
115            elif isinstance(event, BossEncounterComplete):
116                self.boss_encounter_complete_events.add(event)
117            elif isinstance(event, BossEncounterFail):
118                self.boss_encounter_fail_events.add(event)
119            elif isinstance(event, GainCoin):
120                self.gain_coin_events.add(event)
121            elif isinstance(event, BuyUpgrade):
122                self.buy_upgrade_events.add(event)
123            elif isinstance(event, SettingsChange):
124                self.settings_change_events.add(event)
125            elif isinstance(event, KillEnemy):
126                self.kill_enemy_events.add(event)

Creates objects from the json file provided.

Parameters
  • filename: json file with custom schema.
def fail_difficulty_spikes(self) -> dict[int, int]: 
129    def fail_difficulty_spikes(self) -> dict[int, int]:
130        """
131        Output failure rate by stage.
132        
133        :return: dictionary of key stage number and value number of 
134        failures.
135        :rtype: dict[int, int]
136        """
137        difficulty_output = {stage_number: 0 for stage_number in range(1,11)}
138        for event in self.normal_encounter_fail_events:
139            difficulty_output[event.stage_number] += 1
140        for event in self.boss_encounter_fail_events:
141            difficulty_output[event.stage_number] += 1
142        return difficulty_output

Output failure rate by stage.

Returns

dictionary of key stage number and value number of failures.

def get_number_of_session_starts(self) -> int: 
145    def get_number_of_session_starts(self) -> int:
146        """
147        Get the number of start session events.        
148
149        :return: Returns the number of unique session start events.
150        :rtype: int
151        """
152        return len(self.start_session_events)

Get the number of start session events.

Returns

Returns the number of unique session start events.

def get_unique_userIDs(self) -> set[int]: 
155    def get_unique_userIDs(self) -> set[int]:
156        """
157        Returns the set of unique user IDs.
158        
159        :return: set of unique user IDs.
160        :rtype: set[int]
161        """
162        uniqueIDs = set()
163        for event in self.start_session_events:
164            uniqueIDs.add(event.userID)
165        return uniqueIDs

Returns the set of unique user IDs.

Returns

set of unique user IDs.

def count_starts(self, stage_number) -> int: 
168    def count_starts(self, stage_number) -> int:
169        """
170        Counts the number of starts of a given stage.
171        
172        :param stage_number: stage number in question.
173        :return: number of starts of that stage.
174        :rtype: int
175        """
176        start_count = 0
177        if stage_number in [3, 6, 9, 10]:
178            for start_event in self.boss_encounter_start_events:
179                start_count += start_event.stage_number == stage_number
180            return start_count
181        for start_event in self.normal_encounter_start_events:
182            start_count += start_event.stage_number == stage_number
183        return start_count

Counts the number of starts of a given stage.

Parameters
  • stage_number: stage number in question.
Returns

number of starts of that stage.

def count_fails(self, stage_number) -> int: 
186    def count_fails(self, stage_number) -> int:
187        """
188        Counts the number of failures on a given stage.
189        
190        :param stage_number: stage number in question.
191        :return: number of fails at that stage.
192        :rtype: int
193        """
194        fail_count = 0
195        if stage_number in [3, 6, 9, 10]:
196            for fail_event in self.boss_encounter_fail_events:
197                fail_count += fail_event.stage_number == stage_number
198            return fail_count
199        for fail_event in self.normal_encounter_fail_events:
200            fail_count += fail_event.stage_number == stage_number
201        return fail_count

Counts the number of failures on a given stage.

Parameters
  • stage_number: stage number in question.
Returns

number of fails at that stage.

def funnel_view(self) -> dict[int, int]: 
204    def funnel_view(self) -> dict[int, int]:
205        """
206        Output number of players passing a given stage.
207        
208        :return: dictionary of key stage number and value number of 
209        players left.
210        :rtype: dict[int, int]
211        """
212        return {
213            stage_number: 
214            self.count_starts(stage_number) - self.count_fails(stage_number) 
215            for stage_number in range(1,11)
216        }

Output number of players passing a given stage.

Returns

dictionary of key stage number and value number of players left.

def health_per_stage(self, sessionID: int) -> dict[int, int]: 
219    def health_per_stage(self, sessionID: int) -> dict[int, int]:
220        """
221        Output the health a session has per stage.
222        
223        :param sessionID: session stage to check.
224        :type sessionID: int
225        :return: dictionary of key stage number and value player HP
226        remaining.
227        :rtype: dict[int, int]
228        """
229        health_remaining_per_stage = {stage_number: 0 
230                                      for stage_number in range(1,11)}
231        for event in self.normal_encounter_complete_events:
232            if event.sessionID == sessionID:
233                health_remaining_per_stage[
234                    event.stage_number
235                ] = event.player_HP_remaining
236        for event in self.boss_encounter_complete_events:
237            if event.sessionID == sessionID:
238                health_remaining_per_stage[
239                    event.stage_number
240                ] = event.player_HP_remaining
241        return health_remaining_per_stage

Output the health a session has per stage.

Parameters
  • sessionID: session stage to check.
Returns

dictionary of key stage number and value player HP remaining.

def get_difficulty(self, sessionID: int) -> telemetry.core.events.Difficulty: 
244    def get_difficulty(self, sessionID: int) -> Difficulty:
245        """
246        Get the difficulty for a given session.
247        
248        :param sessionID: sessionID of the session to get difficulty of.
249        :type sessionID: int
250        :return: Difficulty value of the session.
251        :rtype: Difficulty
252        """
253        for start_event in self.start_session_events:
254            if start_event.sessionID == sessionID:
255                return start_event.difficulty
256        raise RuntimeError("No session start event for provided " \
257        f"session ID: {sessionID}")

Get the difficulty for a given session.

Parameters
  • sessionID: sessionID of the session to get difficulty of.
Returns

Difficulty value of the session.

def get_sessionIDs_of_difficulty(self, difficulty: telemetry.core.events.Difficulty) -> list[int]: 
260    def get_sessionIDs_of_difficulty(self, difficulty: Difficulty) -> list[int]:
261        """
262        Get the list of all sessionIDs of a given difficulty.
263        
264        :param difficulty: The difficulty level to search for.
265        :type difficulty: Difficulty
266        :return: The list of sessionIDs with the given difficulty.
267        :rtype: list[int]
268        """
269        difficulty_sessionIDs: list[int] = []
270        for start_event in self.start_session_events:
271            if start_event.difficulty == difficulty:
272                difficulty_sessionIDs.append(start_event.sessionID)
273        return difficulty_sessionIDs

Get the list of all sessionIDs of a given difficulty.

Parameters
  • difficulty: The difficulty level to search for.
Returns

The list of sessionIDs with the given difficulty.

def get_health_per_stage_by_difficulty( self, difficulty: telemetry.core.events.Difficulty) -> list[dict[int, int]]: 
276    def get_health_per_stage_by_difficulty(
277            self,
278            difficulty: Difficulty
279    ) -> list[dict[int, int]]:
280        """
281        Returns a health_per_stage dictionary for all sessions with a
282        given difficulty.
283        
284        :param difficulty: Given difficulty.
285        :type difficulty: Difficulty
286        :return: List of health per stage for each session with the
287        specified difficulty level.
288        :rtype: list[dict[int, int]]
289        """
290        health_per_stage = []
291        for sessionID in self.get_sessionIDs_of_difficulty(difficulty):
292            health_per_stage.append(self.health_per_stage(sessionID))
293        return health_per_stage

Returns a health_per_stage dictionary for all sessions with a given difficulty.

Parameters
  • difficulty: Given difficulty.
Returns

List of health per stage for each session with the specified difficulty level.

def compare_health_per_stage_per_difficulty(self) -> dict[telemetry.core.events.Difficulty, list[dict[int, int]]]: 
296    def compare_health_per_stage_per_difficulty(
297            self
298    ) -> dict[Difficulty, list[dict[int, int]]]:
299        """
300        Get a dictionary with difficulty level as the key, and the list 
301        of the health_per_stage objects of all sessions with that 
302        difficulty level as the value.
303        
304        :return: Dictionary of difficulty level to list of all 
305        health_per_stage dictionaries of all sessions with the given
306        difficulty level.
307        :rtype: dict[Difficulty, list[dict[int, int]]]
308        """
309        return {diff: self.get_health_per_stage_by_difficulty(diff) 
310                for diff in Difficulty}

Get a dictionary with difficulty level as the key, and the list of the health_per_stage objects of all sessions with that difficulty level as the value.

Returns

Dictionary of difficulty level to list of all health_per_stage dictionaries of all sessions with the given difficulty level.

def get_coins_gained_per_stage(self, sessionID: int) -> dict[int, int]: 
313    def get_coins_gained_per_stage(
314            self, 
315            sessionID: int
316    ) -> dict[int, int]:
317        """
318        Get the number of coins gained at each stage for a given 
319        session.
320        
321        :param sessionID: sessionID in question.
322        :type sessionID: int
323        :return: Stage number -> coins gained at that stage.
324        :rtype: dict[int, int]
325        """
326        coins_gained_per_stage: dict[int, int] = {i: 0 for i in range(1, 11)}
327        for event in self.gain_coin_events:
328            if event.sessionID == sessionID:
329                coins_gained_per_stage[event.stage_number] += event.coins_gained
330        return coins_gained_per_stage

Get the number of coins gained at each stage for a given session.

Parameters
  • sessionID: sessionID in question.
Returns

Stage number -> coins gained at that stage.

def get_coins_per_stage_by_difficulty( self, difficulty: telemetry.core.events.Difficulty) -> list[dict[int, int]]: 
333    def get_coins_per_stage_by_difficulty(
334            self,
335            difficulty: Difficulty
336    ) -> list[dict[int, int]]:
337        """
338        Returns the coins gained per stage dictionary for all sessions 
339        with a given difficulty.
340        
341        :param difficulty: Given difficulty.
342        :type difficulty: Difficulty
343        :return: List of coins per stage for each session with the
344        specified difficulty level.
345        :rtype: list[dict[int, int]]
346        """
347        coins_per_stage = []
348        for sessionID in self.get_sessionIDs_of_difficulty(difficulty):
349            coins_per_stage.append(self.get_coins_gained_per_stage(sessionID))
350        return coins_per_stage

Returns the coins gained per stage dictionary for all sessions with a given difficulty.

Parameters
  • difficulty: Given difficulty.
Returns

List of coins per stage for each session with the specified difficulty level.

def compare_coins_per_stage_per_difficulty(self) -> dict[telemetry.core.events.Difficulty, list[dict[int, int]]]: 
353    def compare_coins_per_stage_per_difficulty(
354            self
355    ) -> dict[Difficulty, list[dict[int, int]]]:
356        """
357        Get a dictionary with difficulty level as the key, and the list 
358        of the coins gained per stage dictionary objects of all sessions
359        with that difficulty level as the value.
360        
361        :return: Dictionary of difficulty level to list of all 
362        coins gained per stage dictionaries of all sessions with the 
363        given difficulty level.
364        :rtype: dict[Difficulty, list[dict[int, int]]]
365        """
366        return {diff: self.get_coins_per_stage_by_difficulty(diff) 
367                for diff in Difficulty}

Get a dictionary with difficulty level as the key, and the list of the coins gained per stage dictionary objects of all sessions with that difficulty level as the value.

Returns

Dictionary of difficulty level to list of all coins gained per stage dictionaries of all sessions with the given difficulty level.