-
Notifications
You must be signed in to change notification settings - Fork 105
/
schedule.py
155 lines (132 loc) · 5.84 KB
/
schedule.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
import time
from datetime import datetime, timedelta
import statsapi
import data.teams
import debug
from data import status
from data.game import Game
from data.update import UpdateStatus
GAMES_REFRESH_RATE = 6 * 60
class Schedule:
def __init__(self, config):
self.config = config
self.date = self.__parse_today()
self.starttime = time.time()
self.current_idx = 0
self.preferred_over = False
# all games for the day
self.__all_games = []
# the actual (filtered) schedule
self._games = []
self.update(True)
def __parse_today(self):
if self.config.demo_date:
today = datetime.strptime(self.config.demo_date, "%Y-%m-%d")
else:
today = datetime.today()
end_of_day = datetime.strptime(self.config.end_of_day, "%H:%M").replace(
year=today.year, month=today.month, day=today.day
)
if end_of_day > datetime.now():
today -= timedelta(days=1)
return today
def update(self, force=False) -> UpdateStatus:
if force or self.__should_update():
self.date = self.__parse_today()
debug.log("Updating schedule for %s", self.date)
self.starttime = time.time()
try:
self.__all_games = statsapi.schedule(self.date.strftime("%Y-%m-%d"))
except:
debug.exception("Networking error while refreshing schedule")
return UpdateStatus.FAIL
else:
self._games = self.__all_games
if self.config.rotation_only_preferred:
self._games = Schedule.__filter_list_of_games(self.__all_games, self.config.preferred_teams)
if self.config.rotation_only_live:
games = [g for g in self._games if status.is_live(g["status"]) or status.is_fresh(g["status"])]
if games:
self._games = games
return UpdateStatus.SUCCESS
return UpdateStatus.DEFERRED
def __should_update(self):
endtime = time.time()
return endtime - self.starttime >= GAMES_REFRESH_RATE
# offday code
def is_offday_for_preferred_team(self):
if self.config.preferred_teams:
return not any(
data.teams.TEAM_FULL[self.config.preferred_teams[0]] in [game["away_name"], game["home_name"]]
for game in self.__all_games # only care if preferred team is actually in list
)
else:
return True
def is_offday(self):
if self.config.standings_no_games:
return not len(self.__all_games) # care about all MLB
else: # only care if we can't rotate a game
return not len(self._games)
def games_live(self):
return any(status.is_fresh(g["status"]) or (status.is_live(g["status"])) for g in self._games)
def num_games(self):
return len(self._games)
def get_preferred_game(self):
team_index = self._game_index_for_preferred_team()
self.current_idx = team_index
return self.__current_game()
def next_game(self):
# We only need to check the preferred team's game status if we're
# rotating during mid-innings
# TODO verify during a double-header
if (
not self.config.rotation_preferred_team_live_enabled
and self.config.rotation_preferred_team_live_mid_inning
and not self.is_offday_for_preferred_team()
and not self.preferred_over
):
game_index = self._game_index_for_preferred_team()
scheduled_game = self._games[game_index]
preferred_game = Game.from_ID(scheduled_game["game_id"], self.date, scheduled_game["national_broadcasts"])
if preferred_game is not None:
debug.log(
"Preferred Team's Game Status: %s, %s %d",
preferred_game.status(),
preferred_game.inning_state(),
preferred_game.inning_number(),
)
if status.is_live(preferred_game.status()) and not status.is_inning_break(
preferred_game.inning_state()
):
self.current_idx = game_index
debug.log("Moving to preferred game, index: %d", self.current_idx)
return preferred_game
if status.is_complete(preferred_game.status()):
self.preferred_over = True
self.current_idx = self.__next_game_index()
return self.__current_game()
def _game_index_for_preferred_team(self):
if self.config.preferred_teams:
team_name = data.teams.TEAM_FULL[self.config.preferred_teams[0]]
team_index = self.current_idx
team_idxs = [i for i, game in enumerate(self._games) if team_name in [game["away_name"], game["home_name"]]]
if len(team_idxs) > 0:
team_index = next((i for i in team_idxs if status.is_live(self._games[i]["status"])), team_idxs[0],)
return team_index
else:
return self.current_idx
def __next_game_index(self):
counter = self.current_idx + 1
if counter >= len(self._games):
counter = 0
debug.log("Going to game index %d", counter)
return counter
def __current_game(self):
if self._games:
scheduled_game = self._games[self.current_idx]
return Game.from_ID(scheduled_game["game_id"], self.date, scheduled_game["national_broadcasts"])
return None
@staticmethod
def __filter_list_of_games(games, filter_teams):
teams = [data.teams.TEAM_FULL[t] for t in filter_teams]
return list(game for game in games if set([game["away_name"], game["home_name"]]).intersection(set(teams)))