1  from pprint import pprint  # noqa
  2  
  3  from django.test import TransactionTestCase
  4  
  5  from telemetry.models import Coach, Driver, TrackGuide
  6  from telemetry.pitcrew.coach_app import CoachApp
  7  from telemetry.pitcrew.history import History
  8  
  9  from .utils import get_session_df, read_responses, save_responses  # noqa
 10  
 11  
 12  class TestTrackGuideApp(TransactionTestCase):
 13      fixtures = [
 14          "game.json",
 15          "track.json",
 16          "car.json",
 17          "session.json",
 18          "sessiontype.json",
 19          "lap.json",
 20          "fastlap.json",
 21          "fastlapsegment.json",
 22          "driver.json",
 23          "coach.json",
 24          "trackguide.json",
 25          "trackguidenote.json",
 26          "landmark.json",
 27      ]
 28      maxDiff = None
 29  
 30      do_save_responses = True
 31  
 32      def test_no_trackguide(self):
 33          # iRacing / Mazda MX-5 Cup / okayama short
 34  
 35          session_id = "1690362827"
 36          driver = Driver.objects.get(name="durandom")
 37          coach = driver.coach
 38          coach.mode = Coach.MODE_TRACK_GUIDE_APP
 39          coach.save()
 40  
 41          # delete the trackguide
 42          TrackGuide.objects.get(track__name="okayama short", car__name="Mazda MX-5 Cup").delete()
 43  
 44          history = History()
 45  
 46          coach = CoachApp(history, coach)
 47  
 48          session_df = get_session_df(session_id, measurement="fast_laps", bucket="fast_laps")
 49  
 50          row = session_df.iloc[0].to_dict()
 51          topic = row["topic"].replace("Jim", "durandom")
 52          coach.notify(topic, row)
 53          history.init()
 54          history._do_init = False
 55  
 56          captured_responses = []
 57          try:
 58              for index, row in session_df.iterrows():
 59                  row = row.to_dict()
 60                  response = coach.notify(topic, row, row["_time"])
 61                  if response:
 62                      captured_responses.append(response)
 63          except Exception as e:
 64              raise e
 65          finally:
 66              print("stopping history thread")
 67              history.disconnect()
 68  
 69          responses_file = "test_track_guide_app_no_trackguide"
 70          expected_responses = read_responses(responses_file)
 71          if self.do_save_responses:
 72              save_responses(captured_responses, responses_file)
 73  
 74          # pprint(captured_responses, width=200)
 75          self.assertEqual(captured_responses, expected_responses)
 76  
 77      def test_no_data_found(self):
 78          # iRacing / Mazda MX-5 Cup / okayama short
 79  
 80          session_id = "1690362827"
 81          driver = Driver.objects.get(name="durandom")
 82          coach = driver.coach
 83          coach.mode = Coach.MODE_TRACK_GUIDE_APP
 84          coach.save()
 85  
 86          history = History()
 87  
 88          coach = CoachApp(history, coach)
 89  
 90          session_df = get_session_df(session_id, measurement="fast_laps", bucket="fast_laps")
 91  
 92          row = session_df.iloc[0].to_dict()
 93          topic = row["topic"].replace("Jim", "durandom")
 94          topic = row["topic"].replace("Mazda MX-5 Cup", "Ferrari 488 GT3 Evo 2020")
 95          coach.notify(topic, row)
 96          history.init()
 97          history._do_init = False
 98  
 99          captured_responses = []
100          try:
101              for index, row in session_df.iterrows():
102                  row = row.to_dict()
103                  response = coach.notify(topic, row, row["_time"])
104                  if response:
105                      captured_responses.append(response)
106          except Exception as e:
107              raise e
108          finally:
109              print("stopping history thread")
110              history.disconnect()
111  
112          responses_file = "test_track_guide_app_no_data_found"
113          expected_responses = read_responses(responses_file)
114          if self.do_save_responses:
115              save_responses(captured_responses, responses_file)
116  
117          # pprint(captured_responses, width=200)
118          self.assertEqual(captured_responses, expected_responses)
119  
120      def test_track_guide_iracing(self):
121          # iRacing / Mazda MX-5 Cup / okayama short
122  
123          # session by durandom
124          # first 2 laps are slow
125          # 3rd lap has a spin in the first turn
126          # reset to pits
127          # fast out of pits
128          # spin at turn 2
129          # continue at fast pace for one more lap
130          # then slow down after finish line
131          session_id = "1694266648"
132          driver = Driver.objects.get(name="durandom")
133          coach = driver.coach
134          coach.mode = Coach.MODE_TRACK_GUIDE_APP
135          coach.save()
136  
137          history = History()
138  
139          coach = CoachApp(history, coach)
140  
141          session_df = get_session_df(session_id, measurement="laps_cc", bucket="racing")
142  
143          row = session_df.iloc[0].to_dict()
144          topic = row["topic"].replace("Jim", "durandom")
145          coach.notify(topic, row)
146          history.init()
147          history._do_init = False
148  
149          captured_responses = []
150          try:
151              for index, row in session_df.iterrows():
152                  row = row.to_dict()
153                  response = coach.notify(topic, row, row["_time"])
154                  if response:
155                      distance_response = (row["DistanceRoundTrack"], response)
156                      history.log_debug(distance_response)
157                      captured_responses.append(distance_response)
158          except Exception as e:
159              raise e
160          finally:
161              print("stopping history thread")
162              history.disconnect()
163  
164          responses_file = "test_track_guide_app_iracing"
165          expected_responses = read_responses(responses_file)
166          if self.do_save_responses:
167              save_responses(captured_responses, responses_file)
168  
169          # pprint(captured_responses, width=200)
170          self.assertEqual(captured_responses, expected_responses)