telemetry.gui.gui

  1"""
  2Docstring for telemetry.gui.gui
  3
  4This module is responsible for running the Telemetry App GUI.
  5
  6The GUI is implemented using tkinter.
  7
  8This module is responsible for the logic of refreshing plot data.
  9
 10The GUI logic lives in the TelemetryAppGUI class.
 11"""
 12
 13import tkinter as tk
 14from tkinter import ttk, messagebox
 15import seaborn as sns
 16from pathlib import Path
 17
 18from telemetry.core.logic import EventLogicEngine
 19from telemetry.gui.plotting import PlotTab
 20from telemetry.auth.auth import google_login, Role
 21
 22
 23ROOT_DIRECTORY = Path.cwd().parent
 24POLLING_INTERVAL_MS = 3000
 25
 26
 27class GUI_SETTINGS:
 28    """Stores settings for tkinter GUI appearance."""
 29    WINDOW_TITLE = "Telemetry App"   # Title of window.
 30    WINDOW_GEOMETRY = "800x600"      # Size at startup.
 31    WINDOW_MINIMUM_WIDTH = 600       # Minimum width of window.
 32    WINDOW_MINIMUM_HEIGHT = 450      # Minimum height of window. 
 33    FONT_FAMILY = "Arial"
 34    FONT_SIZE = 12
 35    BACKGROUND_COLOR = "#edd68f"
 36
 37
 38class TelemetryAppGUI(tk.Tk):
 39    def __init__(self) -> None:
 40        super().__init__()
 41        self.title(GUI_SETTINGS.WINDOW_TITLE)
 42        self.geometry(GUI_SETTINGS.WINDOW_GEOMETRY)
 43        self.configure(background=GUI_SETTINGS.BACKGROUND_COLOR)
 44        style = ttk.Style(self)
 45        style.theme_use("clam")
 46        self.file_name = ROOT_DIRECTORY / "event_logs" / "telemetry_events.json"
 47        self.logic_engine = EventLogicEngine()
 48        self.authenticated = False
 49        self.current_user_name = None
 50
 51        style.configure(
 52            ".",
 53            font=(GUI_SETTINGS.FONT_FAMILY, GUI_SETTINGS.FONT_SIZE),
 54            padding=6,
 55        )
 56        style.configure(
 57            "TFrame", 
 58            background=GUI_SETTINGS.BACKGROUND_COLOR
 59        )
 60        style.configure(
 61            "TLabel", 
 62            background=GUI_SETTINGS.BACKGROUND_COLOR
 63        )
 64
 65        self.minsize(
 66            width=GUI_SETTINGS.WINDOW_MINIMUM_WIDTH, 
 67            height=GUI_SETTINGS.WINDOW_MINIMUM_HEIGHT
 68        )
 69
 70        self.notebook = ttk.Notebook(self)
 71        self.notebook.grid(row=0, column=0, sticky="nsew")
 72        self.grid_rowconfigure(0, weight=1)
 73        self.grid_columnconfigure(0, weight=1)
 74
 75        self.tab_home = ttk.Frame(self.notebook)
 76        self.tab_funnel = ttk.Frame(self.notebook)
 77        self.tab_spike = ttk.Frame(self.notebook)
 78        self.tab_curves = ttk.Frame(self.notebook)
 79        self.tab_fairness = ttk.Frame(self.notebook)
 80        self.tab_suggestions = ttk.Frame(self.notebook)
 81
 82        self.notebook.add(self.tab_home, text="Home")
 83
 84        self.make_welcome_screen()
 85
 86        sns.set_theme(style="dark", context="notebook")
 87
 88
 89    def make_welcome_screen(self):
 90        self.welcome_label = ttk.Label(
 91            self.tab_home,
 92            text=self.get_personalised_welcome_message(),
 93            justify="center"
 94        )
 95        self.welcome_label.pack(pady=(30, 15))
 96
 97        self.sign_in_button = ttk.Button(
 98            self.tab_home,
 99            text="Sign in with Google",
100            command=self.handle_sign_in
101        )
102        self.sign_in_button.pack(pady=(10, 20))
103
104
105    def get_personalised_welcome_message(self) -> str:
106        return "Welcome to the Telemetry App" if self.current_user_name is None else "Welcome to the Telemetry App, " + self.current_user_name
107
108
109    def handle_sign_in(self):
110        _, self.current_user_name, role = google_login() # TODO: DEAL WITH ROLE RETURN
111        self.authenticated = True
112        AUTHORISED_ROLES = [Role.DESIGNER, Role.DEVELOPER]
113        if role in AUTHORISED_ROLES:
114            self.sign_in_button.pack_forget()
115            self.welcome_label.config(text=self.get_personalised_welcome_message())
116            self.on_authenticated()
117        else:
118            # self.make_welcome_screen()
119            messagebox.showerror(
120                "Authorisation Error", 
121                "Only Designers and Developers may access telemetry data"
122            )
123
124
125    def on_authenticated(self):
126        self.switch_btn_text = tk.StringVar()
127        self.switch_btn_text.set("Change to simulation data")
128
129        switch_simulation_button = ttk.Button(
130            self.tab_home,
131            textvariable=self.switch_btn_text,
132            command=self.toggle_file
133        )
134        switch_simulation_button.pack(pady=(10,20))
135
136        reset_telemetry_button = ttk.Button(
137            self.tab_home,
138            text="Reset Telemetry Data",
139            command=self.reset_telemetry
140        )
141        reset_telemetry_button.pack(pady=(10,20))
142
143        self.notebook.add(self.tab_funnel, text="Funnel view")
144        self.notebook.add(self.tab_spike, text="Difficulty spike")
145        self.notebook.add(self.tab_curves, text="Health")
146        self.notebook.add(self.tab_fairness, text="Coins")
147        self.notebook.add(self.tab_suggestions, text="Suggestions")
148
149        self.tab_spike.rowconfigure(0, weight=1)
150        self.tab_spike.columnconfigure(0, weight=1)
151
152        self.funnel_plot = PlotTab(
153            parent=self.tab_funnel,
154            title="Funnel view",
155            xlabel="Stage",
156            ylabel="Players remaining after round",
157        )
158        self.spike_plot = PlotTab(
159            parent=self.tab_spike,
160            title="Difficulty spikes",
161            xlabel="Stage",
162            ylabel="Number of failures",
163        )
164        self.spike_suggestion = ttk.Label(
165            self.tab_suggestions,
166            text="Suggestion: " + self.generate_spike_suggestion()
167        )
168        self.spike_suggestion.pack(pady=(30, 15)) 
169        self.curves_plot = PlotTab(
170            parent=self.tab_curves,
171            title="HP remaining by stage)",
172            xlabel="Stage",
173            ylabel="Average HP Remaining",
174        )
175        self.fairness_plot = PlotTab(
176            parent=self.tab_fairness,
177            title="Coins gained per stage",
178            xlabel="Stage",
179            ylabel="Coins gained",
180        )
181
182        self.refresh_all()
183        self.do_auto_refresh()
184
185
186    def do_auto_refresh(self, interval_ms=POLLING_INTERVAL_MS):
187        """Use polling to refresh data for plots."""
188        self.refresh_all()
189        self.after(interval_ms, self.do_auto_refresh, interval_ms)
190
191
192    def toggle_file(self):
193        if self.switch_btn_text.get() == "Change to simulation data":
194            self.switch_btn_text.set("Change to telemetry data")
195            self.file_name = ROOT_DIRECTORY / "event_logs" / "simulation_events.json"
196            self.refresh_all()
197        else:
198            self.switch_btn_text.set("Change to simulation data")
199            self.file_name = ROOT_DIRECTORY / "event_logs" / "telemetry_events.json"
200            self.refresh_all()
201
202
203    def reset_telemetry(self):
204        confirmed = messagebox.askyesno(
205        title = "Switch Data Source",
206        message = "Are you sure you want to reset telemetry data? " 
207            + "All existing telemetry data will be lost")
208        if confirmed:
209            with open(ROOT_DIRECTORY / "event_logs" / "telemetry_events.json", 'w') as f:
210                f.write('')
211
212
213    def refresh_all(self):
214        self.refresh_funnel_graph()
215        self.refresh_coins_gained_plots()
216        self.refresh_difficulty_spike_failure_plot()
217        self.refresh_health_plots()
218
219
220    def refresh_funnel_graph(self):
221        """
222        Refreshes the plot of players remaining per stage (referred to
223        as funnel view).
224        """
225        self.logic_engine.categorise_events(self.file_name)
226        funnel_data: dict[int, int] = self.logic_engine.funnel_view()
227        self.funnel_plot.plot_line(
228            funnel_data.keys(), 
229            funnel_data.values(), 
230            label="Players Remaining"
231        )
232
233
234    def refresh_difficulty_spike_failure_plot(self):
235        """
236        Refreshes the plots for difficulty spike in terms of number 
237        of failures per stage.
238        """
239        self.logic_engine.categorise_events(self.file_name)
240        spike_data: dict[int, int] = self.logic_engine.fail_difficulty_spikes()
241        self.spike_plot.plot_line(
242            spike_data.keys(), 
243            spike_data.values(), 
244            label="Difficulty spikes (by failure rate)"
245        )
246        self.spike_suggestion.config(text="Suggestion: " + self.generate_spike_suggestion())
247
248
249    def get_average_dict_of_stage_dicts(
250            self, 
251            per_stage_data: list[dict[int, int]]
252    ) -> dict[int, float]:
253        """
254        Helper function to get the average of a list of dictionaries
255        which map stage numbers to values like HP remaining or coins
256        gained.
257        Example: Given a list of dictionaries of health per stage data,
258        this function would return a dictionary which maps the stage
259        number to the average of health at that stage. 
260        
261        :param per_stage_data: List of dictionaries which represent 
262        "x per stage" data. 
263        :type per_stage_data: list[dict[int, int]]
264        :return: Average of the given data per stage. 
265        :rtype: dict[int, float]
266        """
267        stages = range(1, 11)
268        averages: dict[int, float] = {}
269        for stage in stages:
270            vals = [dictionary.get(stage, 0) 
271                    for dictionary in per_stage_data]
272            averages[stage] = (sum(vals) / len(vals)) if vals else 0.0
273        return averages
274    
275
276    def refresh_health_plots(self):
277        """
278        Refreshes the plots of HP remaining per stage per difficulty.
279        """
280        self.logic_engine.categorise_events(self.file_name)
281        health_by_difficulty = self.logic_engine.compare_health_per_stage_per_difficulty()
282
283        series = []
284        for difficulty, list_of_dicts in health_by_difficulty.items():
285            averages = self.get_average_dict_of_stage_dicts(list_of_dicts)
286            x = averages.keys()
287            y = averages.values()
288            series.append((x, y, str(difficulty.value)))
289
290        self.curves_plot.plot_multi_line(series)
291
292
293    def refresh_coins_gained_plots(self):
294        """
295        Refreshed the plots for average coins gained per stage per
296        difficulty.
297        """
298        self.logic_engine.categorise_events(self.file_name)
299        coins_by_difficulty = self.logic_engine.compare_coins_per_stage_per_difficulty()
300
301        series = []
302        for difficulty, list_of_dicts in coins_by_difficulty.items():
303            averages = self.get_average_dict_of_stage_dicts(list_of_dicts)
304            x = averages.keys()
305            y = averages.values()
306            series.append((x, y, str(difficulty.value)))
307
308        self.fairness_plot.plot_multi_line(series)
309
310
311    def generate_spike_suggestion(self):
312        """
313        Generates a difficulty change suggestion 
314        for difficulty spikes.
315        """
316        stages = ""
317        spikes = self.logic_engine.fail_difficulty_spikes()
318        mean = sum(spikes.values())/len(spikes)
319        for stage in spikes:
320            if spikes[stage] > mean:
321                stages += str(stage) + ", "
322        if stages != "":
323            return "High failure rate in " + stages + " consider increasing lives by 2."
324        return "No suggestions available"