LadderTimer.check_racer_finish()   A
last analyzed

Complexity

Conditions 4

Size

Total Lines 12
Code Lines 11

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 4
eloc 11
nop 1
dl 0
loc 12
rs 9.85
c 0
b 0
f 0
1
import asyncio
2
import logging
3
from datetime import datetime, timedelta
4
import pytz
5
from typing import List
6
from threading import Thread
7
8
from helpers import timer_to_str
9
import clients.ladder_client as ladder_client
10
from models.ladder import Flag, Racer, ScheduleItem, Season, Standings
11
12
13
def str_to_timer(str_timer):
14
    return timedelta(
15
        hours=float(str_timer[0:1]), minutes=float(str_timer[2:4]),
16
        seconds=float(str_timer[5:7]))
17
18
19
class LadderTimer:
20
    logger: logging.Logger = None
21
    enabled: bool = False
22
    source_name = ""
23
    timer_text = ""
24
    racer_id: int = 0
25
    racer_name: str = ""
26
    stats_source: str = ""
27
    stats: Standings = None
28
    season_for_stats: int = 0
29
    mode_for_stats: int = 0
30
    show_season_name: bool = False
31
    show_mode_name: bool = False
32
    show_rating: bool = False
33
    show_rank: bool = False
34
    show_change: bool = False
35
    show_win_loss_tie: bool = False
36
    pre_color: int = 0xFFFFFF
37
    racing_color: int = 0xFFFFFF
38
    winner_color: int = 0xFF0000
39
    loser_color: int = 0x00FF00
40
    ff_color: int = 0x00FF00
41
    started_at: datetime = None
42
    finish_time: timedelta = None
43
    result: str = ""
44
    next_race: ScheduleItem = None
45
    active_racers: List[Racer] = None
46
    current_season: Season = None
47
    flags: List[Flag] = []
48
    all_seasons: List[Season] = []
49
    schedule: List[ScheduleItem] = []
50
    last_timer_update: datetime = None
51
    decimals: bool = True
52
53
    @staticmethod
54
    def ladder_timezone():
55
        return pytz.timezone('US/Eastern')
56
57
    def __init__(self, logger: logging.Logger, racer_id: int = None):
58
        self.logger = logger
59
        if racer_id is not None:
60
            self.racer_id = racer_id
61
62
        ladder_update_t = Thread(target=self.ladder_update_thread)
63
        ladder_update_t.daemon = True
64
        ladder_update_t.start()
65
66
    def ladder_update_thread(self):
67
        self.logger.debug("starting ladder race update")
68
        loop = asyncio.new_event_loop()
69
        loop.run_until_complete(self.race_updater())
70
        loop.run_forever()
71
72
    async def race_updater(self):
73
        await self.update_ladder_status()
74
        while True:
75
            if not self.enabled or self.finish_time is not None:
76
                await asyncio.sleep(5.0)
77
            else:
78
                now = datetime.now(self.ladder_timezone())
79
                if self.started_at is not None and now > self.started_at:
80
                    self.logger.debug("checking if racer finished")
81
                    await self.check_racer_finish()
82
                await asyncio.sleep(5.0)
83
84
    async def update_ladder_status(self):
85
        self.started_at = None
86
        self.finish_time = None
87
        self.last_timer_update = None
88
        self.next_race = None
89
        self.schedule = None
90
        await self.update_active_racers()
91
        await self.update_flags()
92
        await self.update_season()
93
        await self.update_schedule(self.current_season.season_id)
94
        await self.update_timer()
95
        await self.update_stats()
96
97
    async def check_racer_finish(self):
98
        if ladder_client.is_currently_racing(self.racer_id):
99
            return
100
        racer_history = ladder_client.get_racer_history(self.racer_id)
101
        if racer_history[-1].race_id == self.next_race.race_id:
102
            if racer_history[-1].FinishTime == "FF":
103
                self.result = "FF"
104
                self.finish_time = timedelta(seconds=0)
105
                self.logger.debug("racer forfeited")
106
            else:
107
                self.finish_time = str_to_timer(racer_history[-1].FinishTime)
108
                self.result = racer_history[-1].Result
109
110
    def get_timer_text(self):
111
        if self.finish_time:
112
            if self.result == "W":
113
                return (
114
                    self.winner_color,
115
                    timer_to_str(self.finish_time, self.decimals)
116
                )
117
            elif self.result == "L":
118
                return (
119
                    self.loser_color,
120
                    timer_to_str(self.finish_time, self.decimals)
121
                )
122
            elif self.result == "FF":
123
                return self.ff_color, "-:--:--.-"
124
        current_timer = timedelta(seconds=0)
125
        color = self.pre_color
126
        now = datetime.now(self.ladder_timezone())
127
        if self.started_at is None:
128
            current_timer = now - self.next_race.StartTime
129
            if (abs(current_timer.total_seconds()) < 600.0):
130
                loop = asyncio.new_event_loop()
131
                loop.run_until_complete(self.update_timer())
132
                loop.close()
133
            color = self.pre_color
134
        else:
135
            current_timer = now - self.started_at
136
            if current_timer.total_seconds() > 0:
137
                color = self.racing_color
138
        return color, timer_to_str(current_timer, self.decimals)
139
140
    def get_stats(self):
141
        stats = ""
142
        if self.enabled:
143
            if self.show_season_name:
144
                stats = stats + self.stats.Season + " "
145
            if self.show_mode_name:
146
                stats = stats + self.stats.Mode + " "
147
            if self.show_rank:
148
                stats = stats + "#" + str(self.stats.Rank) + " "
149
            if self.show_rating:
150
                stats = stats + str(self.stats.Rating) + " "
151
            if self.show_change:
152
                stats = stats + str(self.stats.Change) + " "
153
            if self.show_win_loss_tie:
154
                stats = (
155
                    stats + str(self.stats.Wins) + "/" +
156
                    str(self.stats.Losses) + "/" + str(self.stats.Ties)
157
                )
158
        return stats
159
160
    def update_settings(self, racer_name: str):
161
        if racer_name is None or racer_name == "":
162
            return False
163
        racer_id = self.get_racer_id(racer_name)
164
        if racer_id == 0:
165
            return False
166
        else:
167
            self.racer_id = racer_id
168
            self.racer_name = racer_name
169
            loop = asyncio.new_event_loop()
170
            loop.run_until_complete(self.update_stats())
171
            loop.close()
172
            return True
173
174
    def get_racer_id(self, racer_name: str) -> Racer:
175
        loop = asyncio.new_event_loop()
176
        loop.run_until_complete(self.update_active_racers())
177
        loop.close()
178
        for racer in self.active_racers:
179
            if racer.RacerName.lower() == racer_name.lower():
180
                return racer.racer_id
181
        return 0
182
183
    async def update_active_racers(self):
184
        if self.active_racers is None or self.active_racers == []:
185
            self.active_racers = ladder_client.get_active_racers()
186
187
    async def update_season(self):
188
        if self.current_season is None:
189
            self.all_seasons = ladder_client.get_seasons()
190
            if self.all_seasons is not None:
191
                self.current_season = (
192
                    next(filter(
193
                        lambda x: x.IsCurrentSeason, self.all_seasons
194
                    ), None)
195
                )
196
        self.logger.info(f"current_season = {self.current_season}")
197
198
    async def update_flags(self):
199
        self.flags = ladder_client.get_flags()
200
201
    async def update_schedule(self, season_id: int):
202
        if self.schedule is None or self.schedule == []:
203
            self.schedule = ladder_client.get_schedule(season_id)
204
            self.next_race = (
205
                next(x for x in self.schedule if not x.HasCompleted)
206
            )
207
            self.logger.info(f"next_race = {self.next_race}")
208
209
    async def update_timer(self):
210
        if (
211
            self.last_timer_update is not None and
212
            (datetime.now() - self.last_timer_update) < timedelta(seconds=20)
213
        ):
214
            return
215
        self.logger.debug("calling get_current_race_time")
216
        str_timer = ladder_client.get_current_race_time()
217
        self.last_timer_update = datetime.now()
218
        self.logger.info(f"str_timer= {str_timer}")
219
        timer_sign = 1.0
220
        if str_timer[0] == '-':
221
            timer_sign = -1.0
222
            str_timer = str_timer[1:]
223
        timer = str_to_timer(str_timer)
224
        timer = timer * timer_sign
225
        if timer == timedelta(seconds=0):
226
            self.started_at = None
227
        else:
228
            self.started_at = datetime.now(self.ladder_timezone()) - timer
229
230
    async def update_stats(self):
231
        if self.season_for_stats == -1:
232
            self.season_for_stats = self.current_season.season_id
233
        if self.mode_for_stats == -1:
234
            self.mode_for_stats = self.get_mode_id_from_name(
235
                self.next_race.Mode)
236
        self.logger.info(f"season_for_stats: {self.season_for_stats}")
237
        self.logger.info(f"mode_for_stats: {self.mode_for_stats}")
238
        standings = ladder_client.get_standings(
239
            self.season_for_stats, self.mode_for_stats)
240
        for stats in standings:
241
            if stats.RacerName.lower() == self.racer_name.lower():
242
                self.stats = stats
243
                return
244
        # default standings
245
        self.stats = Standings(
246
            self.racer_name, self.current_season.SeasonName,
247
            self.next_race.Mode, 1600, 0, 0, 0, 0, 0)
248
249
    def get_mode_id_from_name(self, mode_name: str):
250
        for flag in self.flags:
251
            if flag.Mode == mode_name:
252
                return flag.flag_id
253