test_session.py
1 from pprint import pprint 2 3 import django.utils.timezone 4 from django.db import IntegrityError 5 from django.test import TestCase 6 7 from telemetry.models import Car, Driver, Game 8 from telemetry.models import Session as SessionModel 9 from telemetry.models import SessionType, Track 10 from telemetry.pitcrew.firehose import Firehose 11 from telemetry.pitcrew.session import Lap, Session 12 13 from .utils import get_session_df 14 15 16 class TestSession(TestCase): 17 def _test_session_firehose(self, session_id, measurement="fast_laps", bucket="fast_laps"): 18 session_df = get_session_df(session_id, measurement=measurement, bucket=bucket) 19 20 firehose = Firehose() 21 22 for index, row in session_df.iterrows(): 23 # convert row to dict 24 row = row.to_dict() 25 now = row["_time"] 26 firehose.notify(row["topic"], row, now) 27 if index == 0: 28 session = firehose.sessions[row["topic"]] 29 30 pprint(session.laps) 31 return session 32 33 def _test_session(self, session_id, measurement="fast_laps", bucket="fast_laps"): 34 session_df = get_session_df(session_id, measurement=measurement, bucket=bucket) 35 36 # Create an instance of the Session class 37 test_session = Session(666) 38 39 for index, row in session_df.iterrows(): 40 # convert row to dict 41 row = row.to_dict() 42 now = row["_time"] 43 test_session.signal(row, now) 44 45 pprint(test_session.laps) 46 return test_session 47 48 def _assert_laps(self, test_session, expected_laps): 49 # Iterate over the expected_laps dictionary and compare to the test_session.laps 50 for lap_number, expected_lap in expected_laps.items(): 51 lap = test_session.laps[lap_number] 52 53 self.assertEqual(lap.number, expected_lap.number) 54 self.assertEqual(lap.time, expected_lap.time) 55 self.assertEqual(lap.valid, expected_lap.valid) 56 self.assertEqual(lap.finished, expected_lap.finished) 57 self.assertAlmostEqual(int(lap.length), int(expected_lap.length), places=0) 58 59 if lap.time != -1: 60 # the difference between lap.end and lap.start should be equal to lap.time 61 time_delta = lap.end - lap.start 62 self.assertAlmostEqual(time_delta.total_seconds(), expected_lap.time, places=0) 63 64 def test_iracing(self): 65 # measurement = "fast_laps" 66 # bucket = "fast_laps" 67 # start = "-10y" 68 session_id = "1681021274" 69 70 # For this session the following laps are valid: 71 # 1 valid: during outlap lap_number is 1, remains 1 on crossing finish line 72 # 2 invalid: penalty during lap, reset to pits, lap_number is remains 2 on outlap 73 # 3 valid: lap_number changes to 3 on crossing finish line 74 # 4 valid: lap_number changes to 4 on crossing finish line 75 # 5 valid: lap_number changes to 5 on crossing finish line 76 # 6 invalid: penalty during lap, no reset to pits 77 # 7 invalid: lap_number changes to 7 on crossing finish line, 78 # no penalty during lap, but "PreviousLapWasValid" is false 79 # 8 invalid: penalty during lap, reset to pits, lap_number is remains 8 on outlap 80 # 9 invalid: penalty during lap, no reset to pits 81 # 10 invalid: penalty during lap, eset to pits 82 # a lap time of -1 indicates an outlap 83 84 expected_laps = { 85 1: Lap(1, time=100.818, valid=True, length=4409), 86 2: Lap(2, time=-1, valid=False, length=4408), 87 3: Lap(3, time=101.0466, valid=True, length=4408), 88 4: Lap(4, time=100.823, valid=True, length=4408), 89 5: Lap(5, time=99.4026, valid=True, length=4410), 90 6: Lap(6, time=107.9166, valid=False, length=4410), 91 7: Lap(7, time=99.0674, valid=False, length=4410), 92 8: Lap(8, time=-1, valid=False, length=4409), 93 9: Lap(9, time=101.7361, valid=False, length=4409), 94 10: Lap(10, time=-1, valid=False, length=1890), 95 } 96 for lap_number, expected_lap in expected_laps.items(): 97 expected_lap.finished = True 98 expected_laps[10].finished = False 99 session = self._test_session(session_id) 100 self._assert_laps(session, expected_laps) 101 102 def test_ac(self): 103 # measurement = "fast_laps" 104 # bucket = "fast_laps" 105 # start = "-10y" 106 session_id = "1673613558" 107 108 # For this session the following laps are valid: 109 expected_laps = { 110 2: Lap(2, time=63.79, valid=False, length=2338.449), 111 3: Lap(3, time=62.4660034, valid=True, length=2341.24487), 112 4: Lap(4, time=64.097, valid=False, length=2339.07666), 113 5: Lap(5, time=61.833, valid=True, length=2340.87329), 114 6: Lap(6, time=70.983, valid=False, length=2342.41357), 115 7: Lap(7, time=61.465, valid=True, length=2337.27051), 116 8: Lap(8, time=67.749, valid=True, length=2337.886), 117 9: Lap(9, time=61.703, valid=True, length=2338.9856), 118 10: Lap(10, time=73.402, valid=False, length=2341.15625), 119 11: Lap(11, time=85.821, valid=False, length=2340.723), 120 # 12: Lap(12, time=169.38, valid=False, length=2338.358), # during this lap the game was paused 121 # 12: Lap(12, time=174.319331, valid=False, length=2338.358), 122 13: Lap(13, time=64.435, valid=False, length=2337.65625), 123 14: Lap(14, time=62.7619972, valid=False, length=2341.5625), 124 15: Lap(15, time=77.178, valid=False, length=2341.84863), 125 16: Lap(16, time=79.708, valid=False, length=2342.23486), 126 17: Lap(17, time=-1.0, valid=True, length=33.61092), 127 } 128 for lap_number, expected_lap in expected_laps.items(): 129 expected_lap.finished = True 130 expected_laps[17].finished = False 131 132 session = self._test_session(session_id) 133 self._assert_laps(session, expected_laps) 134 135 def test_rbr(self): 136 # measurement = "fast_laps" 137 # bucket = "fast_laps" 138 # start = "-10y" 139 session_id = "1703706617" 140 141 session = self._test_session_firehose(session_id, measurement="laps_cc", bucket="racing") 142 143 # lap time: 3:30.60 144 # at 1:00 pressed paused 145 # at 1:30 pressed call for help 146 # For this session the following laps are valid: 147 expected_lap = Lap(0, time=210.600174, valid=True, length=4834.398, finished=True) 148 lap = session.laps[0] 149 self.assertEqual(lap.number, expected_lap.number) 150 self.assertEqual(lap.time, expected_lap.time) 151 self.assertEqual(lap.valid, expected_lap.valid) 152 self.assertEqual(lap.finished, expected_lap.finished) 153 self.assertAlmostEqual(int(lap.length), int(expected_lap.length), places=0) 154 155 def test_car_class(self): 156 session_id = "1692140843" 157 158 session = self._test_session_firehose(session_id) 159 160 self.assertEqual(session.car_class, "ARC_CAMERO") 161 162 def test_telemetry_invalid(self): 163 # 2390.0: LapTimePrevious: None -> None 164 # 2390.0: CurrentLapIsValid: None -> None 165 # 2390.0: PreviousLapWasValid: None -> None 166 # these values are always None in that session 167 session_id = "1672395579" 168 session = self._test_session(session_id) 169 self.assertEqual(session.laps, {}) 170 171 def test_telemetry_missing_fields(self): 172 # last lap CurrentLapIsValid is None 173 session_id = "1680321341" 174 session = self._test_session(session_id) 175 176 expected_laps = { 177 11: Lap(11, time=-1, length=82, valid=True, finished=False), 178 } 179 self._assert_laps(session, expected_laps) 180 181 def test_duplicate_lap(self): 182 # create 2 laps with the same number 183 game = Game.objects.create(name="test_game") 184 track = Track.objects.create(name="test_track", game=game) 185 car = Car.objects.create(name="test_car", game=game) 186 driver = Driver.objects.create(name="test_driver") 187 session_type = SessionType.objects.create(type="test_session_type") 188 session = SessionModel.objects.create(session_id=666, driver=driver, game=game, session_type=session_type) 189 190 now = django.utils.timezone.now() 191 192 session.laps.create(number=1, car=car, track=track, start=now) 193 try: 194 session.laps.create(number=2, car=car, track=track, start=now) 195 except IntegrityError as e: 196 self.assertEqual( 197 e.args[0], 198 "UNIQUE constraint failed: telemetry_lap.session_id, telemetry_lap.start", 199 ) 200 except Exception as e: 201 self.fail(f"Unexpected exception: {e}")