telemetry.core.parsing

  1"""
  2Docstring for telemetry.core.parsing
  3
  4This module contains the logic for parsing game output json files. This
  5module is responsible for reading the event log JSON files, and
  6instantiating the ValidEvent object which corresponds to the event 
  7parsed.
  8"""
  9
 10import json
 11from typing import TypeAlias
 12from datetime import datetime
 13from pathlib import Path
 14
 15from telemetry.core.events import (
 16    KillEnemy,
 17    EventParameter,
 18    BossEncounterFail,
 19    GainCoin,
 20    EventType,
 21    StartSession,
 22    BuyUpgrade,
 23    NormalEncounterStart,
 24    EndSession,
 25    NormalEncounterComplete,
 26    SettingsChange,
 27    BossEncounterComplete,
 28    NormalEncounterFail,
 29    BossEncounterStart,
 30)
 31
 32
 33ValidEvent: TypeAlias = (
 34    StartSession
 35    | NormalEncounterStart
 36    | NormalEncounterComplete
 37    | NormalEncounterFail
 38    | BossEncounterStart
 39    | BossEncounterComplete
 40    | BossEncounterFail
 41    | GainCoin
 42    | BuyUpgrade
 43    | EndSession
 44    | SettingsChange
 45    | KillEnemy
 46)
 47
 48    
 49def convert_time(time_string: str) -> datetime:
 50    """
 51    :param time_string: Custom formatted string of timestamp.
 52    :type time_string: str
 53    :return: Datetime version of time_string.
 54    :rtype: datetime
 55    """
 56    return datetime.strptime(time_string, "%Y/%m/%d/%H/%M/%S")
 57
 58
 59def get_file(filename: Path) -> list[dict]:
 60    """
 61    This function parses a json file and returns a list of json objects.
 62    
 63    :param filename: filename for json to be parsed.
 64    :type filename: Path
 65    :return: List of json objects.
 66    :rtype: list[dict[str, Any]]
 67    """
 68    try: 
 69        with open(filename, 'r') as f:
 70            return json.load(f)
 71    except FileNotFoundError:
 72        raise FileNotFoundError(f"Could not find file: {filename}")
 73    except json.JSONDecodeError:
 74        raise RuntimeError(f"Could not parse - invalid json.")
 75
 76
 77def parse_file(filename: Path) -> list[ValidEvent]:
 78    """
 79    Creates a list of ValidEvent objects from a json file.
 80    
 81    :param filename: filename for json to be parsed.
 82    :type filename: Path
 83    :return: List of valid event objects.
 84    :rtype: list[ValidEvent]
 85    """
 86    events = get_file(filename)
 87    event_objects = []
 88    for event in events:
 89        event_objects.append(parse_event(event))
 90    return event_objects
 91
 92
 93def parse_event(event: dict) -> ValidEvent:
 94    """
 95    Creates a ValidEvent object from a json object.
 96    
 97    :param event: json object for a single event.
 98    :type event: dict
 99    :return: Valid event object translated from the json object.
100    :rtype: ValidEvent
101    :raises RuntimeError: If an event of an unknown type is found, or 
102    a required field is missing.
103    """
104    try:
105        event_type = event[EventParameter.EVENT_TYPE]
106    except KeyError:
107        raise RuntimeError("Event must have an \"event\" field")
108    
109    try:
110        match event_type:
111            case EventType.START_SESSION:
112                return StartSession(
113                    event[EventParameter.USER_ID],
114                    event[EventParameter.SESSION_ID],
115                    convert_time(event[EventParameter.TIMESTAMP]),
116                    event[EventParameter.DIFFICULTY]
117                )
118            case EventType.END_SESSION:
119                return EndSession(
120                    event[EventParameter.USER_ID],
121                    event[EventParameter.SESSION_ID],
122                    convert_time(event[EventParameter.TIMESTAMP])
123                )
124            case EventType.NORMAL_ENCOUNTER_START:
125                return NormalEncounterStart(
126                    event[EventParameter.USER_ID],
127                    event[EventParameter.SESSION_ID],
128                    convert_time(event[EventParameter.TIMESTAMP]),
129                    event[EventParameter.ENCOUNTER],
130                    event[EventParameter.DIFFICULTY],
131                    event[EventParameter.STAGE_NUMBER]
132                )
133            case EventType.NORMAL_ENCOUNTER_COMPLETE:
134                return NormalEncounterComplete(
135                    event[EventParameter.USER_ID],
136                    event[EventParameter.SESSION_ID],
137                    convert_time(event[EventParameter.TIMESTAMP]),
138                    event[EventParameter.ENCOUNTER],
139                    event[EventParameter.DIFFICULTY],
140                    event[EventParameter.STAGE_NUMBER],
141                    event[EventParameter.PLAYER_HP_REMAINING]
142                )
143            case EventType.NORMAL_ENCOUNTER_FAIL:
144                return NormalEncounterFail(
145                    event[EventParameter.USER_ID],
146                    event[EventParameter.SESSION_ID],
147                    convert_time(event[EventParameter.TIMESTAMP]),
148                    event[EventParameter.ENCOUNTER],
149                    event[EventParameter.DIFFICULTY],
150                    event[EventParameter.STAGE_NUMBER],
151                    event[EventParameter.LIVES_LEFT]
152                )
153            case EventType.BOSS_ENCOUNTER_START:
154                return BossEncounterStart(
155                    event[EventParameter.USER_ID],
156                    event[EventParameter.SESSION_ID],
157                    convert_time(event[EventParameter.TIMESTAMP]),
158                    event[EventParameter.ENCOUNTER],
159                    event[EventParameter.DIFFICULTY],
160                    event[EventParameter.STAGE_NUMBER]
161                )
162            case EventType.BOSS_ENCOUNTER_COMPLETE:
163                return BossEncounterComplete(
164                    event[EventParameter.USER_ID],
165                    event[EventParameter.SESSION_ID],
166                    convert_time(event[EventParameter.TIMESTAMP]),
167                    event[EventParameter.ENCOUNTER],
168                    event[EventParameter.DIFFICULTY],
169                    event[EventParameter.STAGE_NUMBER],
170                    event[EventParameter.PLAYER_HP_REMAINING]
171                )
172            case EventType.BOSS_ENCOUNTER_FAIL:
173                return BossEncounterFail(
174                    event[EventParameter.USER_ID],
175                    event[EventParameter.SESSION_ID],
176                    convert_time(event[EventParameter.TIMESTAMP]),
177                    event[EventParameter.ENCOUNTER],
178                    event[EventParameter.DIFFICULTY],
179                    event[EventParameter.STAGE_NUMBER],
180                    event[EventParameter.LIVES_LEFT]
181                )
182            case EventType.GAIN_COIN:
183                return GainCoin(
184                    event[EventParameter.USER_ID],
185                    event[EventParameter.SESSION_ID],
186                    convert_time(event[EventParameter.TIMESTAMP]),
187                    event[EventParameter.ENCOUNTER],
188                    event[EventParameter.DIFFICULTY],
189                    event[EventParameter.STAGE_NUMBER],
190                    event[EventParameter.COINS_GAINED]
191                )
192            case EventType.BUY_UPGRADE:
193                return BuyUpgrade(
194                    event[EventParameter.USER_ID],
195                    event[EventParameter.SESSION_ID],
196                    convert_time(event[EventParameter.TIMESTAMP]),
197                    event[EventParameter.STAGE_NUMBER],
198                    event[EventParameter.COINS_SPENT],
199                    event[EventParameter.UPGRADE_BOUGHT]
200                )
201            case EventType.SETTINGS_CHANGE:
202                return SettingsChange(
203                    event[EventParameter.USER_ID],
204                    convert_time(event[EventParameter.TIMESTAMP]),
205                    event[EventParameter.SETTING],
206                    event[EventParameter.SETTING_VALUE]
207                )
208            case EventType.KILL_ENEMY:
209                return KillEnemy(
210                    event[EventParameter.USER_ID],
211                    event[EventParameter.SESSION_ID],
212                    convert_time(event[EventParameter.TIMESTAMP]),
213                    event[EventParameter.ENCOUNTER],
214                    event[EventParameter.DIFFICULTY],
215                    event[EventParameter.STAGE_NUMBER],
216                    event[EventParameter.ENEMY_TYPE]
217                )
218            case _:
219                raise RuntimeError(
220                    f"Unexpected event type: {event_type}"
221                )
222    except KeyError as e:
223        raise RuntimeError(
224            f"An event of type {event_type} is missing the field {e}"
225        )