1
|
|
|
from dataclasses import dataclass |
2
|
|
|
from typing import Any, Optional, List |
3
|
|
|
from datetime import datetime, timedelta |
4
|
|
|
from models import ( |
5
|
|
|
from_str, from_union, from_none, from_timedelta, |
6
|
|
|
from_int, from_bool, from_datetime, from_list |
7
|
|
|
) |
8
|
|
|
from models.user import User |
9
|
|
|
|
10
|
|
|
|
11
|
|
|
@dataclass |
12
|
|
|
class RaceCategory: |
13
|
|
|
name: str |
14
|
|
|
short_name: str |
15
|
|
|
slug: str |
16
|
|
|
url: str |
17
|
|
|
data_url: str |
18
|
|
|
image: str |
19
|
|
|
|
20
|
|
|
@staticmethod |
21
|
|
|
def from_dict(obj: Any) -> 'RaceCategory': |
22
|
|
|
if not isinstance(obj, dict): |
23
|
|
|
return None |
24
|
|
|
name = from_str(obj.get("name")) |
25
|
|
|
short_name = from_str(obj.get("short_name")) |
26
|
|
|
slug = from_str(obj.get("slug")) |
27
|
|
|
url = from_str(obj.get("url")) |
28
|
|
|
data_url = from_str(obj.get("data_url")) |
29
|
|
|
image = from_union([from_str, from_none], obj.get("image")) |
30
|
|
|
return RaceCategory(name, short_name, slug, url, data_url, image) |
31
|
|
|
|
32
|
|
|
|
33
|
|
|
@dataclass |
34
|
|
|
class Status: |
35
|
|
|
value: str |
36
|
|
|
verbose_value: str |
37
|
|
|
help_text: str |
38
|
|
|
|
39
|
|
|
@staticmethod |
40
|
|
|
def from_dict(obj: Any) -> 'Status': |
41
|
|
|
if not isinstance(obj, dict): |
42
|
|
|
return None |
43
|
|
|
value = from_str(obj.get("value")) |
44
|
|
|
verbose_value = from_str(obj.get("verbose_value")) |
45
|
|
|
help_text = from_str(obj.get("help_text")) |
46
|
|
|
return Status(value, verbose_value, help_text) |
47
|
|
|
|
48
|
|
|
|
49
|
|
|
@dataclass |
50
|
|
|
class Entrant: |
51
|
|
|
user: User |
52
|
|
|
status: Status |
53
|
|
|
has_comment: bool |
54
|
|
|
stream_live: bool |
55
|
|
|
stream_override: bool |
56
|
|
|
actions: List[str] |
57
|
|
|
score: Optional[int] = None |
58
|
|
|
finish_time: Optional[timedelta] = None |
59
|
|
|
finished_at: Optional[datetime] = None |
60
|
|
|
place: Optional[int] = None |
61
|
|
|
place_ordinal: Optional[str] = None |
62
|
|
|
score_change: Optional[int] = None |
63
|
|
|
comment: Optional[str] = None |
64
|
|
|
|
65
|
|
|
@staticmethod |
66
|
|
|
def from_dict(obj: Any) -> 'Entrant': |
67
|
|
|
if not isinstance(obj, dict): |
68
|
|
|
return None |
69
|
|
|
user = User.from_dict(obj.get("user")) |
70
|
|
|
status = Status.from_dict(obj.get("status")) |
71
|
|
|
finish_time = from_union( |
72
|
|
|
[from_timedelta, from_none], obj.get("finish_time")) |
73
|
|
|
finished_at = from_union( |
74
|
|
|
[from_datetime, from_none], obj.get("finished_at")) |
75
|
|
|
place = from_union([from_int, from_none], obj.get("place")) |
76
|
|
|
place_ordinal = from_union( |
77
|
|
|
[from_str, from_none], obj.get("place_ordinal")) |
78
|
|
|
score_change = from_union( |
79
|
|
|
[from_int, from_none], obj.get("score_change")) |
80
|
|
|
comment = from_union([from_str, from_none], obj.get("comment")) |
81
|
|
|
has_comment = from_bool(obj.get("has_comment")) |
82
|
|
|
stream_live = from_bool(obj.get("stream_live")) |
83
|
|
|
stream_override = from_bool(obj.get("stream_override")) |
84
|
|
|
actions = from_list(from_str, obj.get("actions")) |
85
|
|
|
score = from_union([from_int, from_none], obj.get("score")) |
86
|
|
|
return Entrant( |
87
|
|
|
user=user, status=status, finish_time=finish_time, |
88
|
|
|
finished_at=finished_at, place=place, place_ordinal=place_ordinal, |
89
|
|
|
score_change=score_change, comment=comment, |
90
|
|
|
has_comment=has_comment, stream_live=stream_live, |
91
|
|
|
stream_override=stream_override, actions=actions, score=score |
92
|
|
|
) |
93
|
|
|
|
94
|
|
|
|
95
|
|
|
@dataclass |
96
|
|
|
class Goal: |
97
|
|
|
name: str |
98
|
|
|
custom: bool |
99
|
|
|
|
100
|
|
|
@staticmethod |
101
|
|
|
def from_dict(obj: Any) -> 'Goal': |
102
|
|
|
if not isinstance(obj, dict): |
103
|
|
|
return None |
104
|
|
|
name = from_str(obj.get("name")) |
105
|
|
|
custom = from_bool(obj.get("custom")) |
106
|
|
|
return Goal(name, custom) |
107
|
|
|
|
108
|
|
|
|
109
|
|
|
@dataclass |
110
|
|
|
class Race: |
111
|
|
|
name: str |
112
|
|
|
status: Status |
113
|
|
|
url: str |
114
|
|
|
data_url: str |
115
|
|
|
category: RaceCategory |
116
|
|
|
goal: Goal |
117
|
|
|
info: str |
118
|
|
|
entrants_count: int |
119
|
|
|
entrants_count_inactive: int |
120
|
|
|
opened_at: datetime |
121
|
|
|
time_limit: timedelta |
122
|
|
|
slug: Optional[str] = None |
123
|
|
|
entrants: Optional[List[Entrant]] = None |
124
|
|
|
version: Optional[int] = None |
125
|
|
|
started_at: Optional[datetime] = None |
126
|
|
|
ended_at: Optional[datetime] = None |
127
|
|
|
cancelled_at: Optional[datetime] = None |
128
|
|
|
recorded_by: Optional[User] = None |
129
|
|
|
start_delay: Optional[timedelta] = None |
130
|
|
|
streaming_required: Optional[bool] = None |
131
|
|
|
auto_start: Optional[bool] = None |
132
|
|
|
opened_by: Optional[User] = None |
133
|
|
|
monitors: Optional[List[User]] = None |
134
|
|
|
recordable: Optional[bool] = None |
135
|
|
|
recorded: Optional[bool] = None |
136
|
|
|
allow_comments: Optional[bool] = None |
137
|
|
|
hide_comments: Optional[bool] = None |
138
|
|
|
allow_midrace_chat: Optional[bool] = None |
139
|
|
|
allow_non_entrant_chat: Optional[bool] = None |
140
|
|
|
chat_message_delay: Optional[str] = None |
141
|
|
|
unlisted: Optional[bool] = None |
142
|
|
|
entrants_count_finished: Optional[int] = None |
143
|
|
|
websocket_url: Optional[str] = None |
144
|
|
|
websocket_bot_url: Optional[str] = None |
145
|
|
|
websocket_oauth_url: Optional[str] = None |
146
|
|
|
|
147
|
|
|
def get_entrant_by_name(self, full_name: str) -> Entrant: |
148
|
|
|
entrant = next( |
149
|
|
|
(x for x in self.entrants if x.user.full_name == full_name), None) |
|
|
|
|
150
|
|
|
return entrant |
151
|
|
|
|
152
|
|
|
def get_entrant_by_place(self, place: int) -> Entrant: |
153
|
|
|
entrant = next( |
154
|
|
|
(x for x in self.entrants if x.place == place), None) |
|
|
|
|
155
|
|
|
return entrant |
156
|
|
|
|
157
|
|
|
@staticmethod |
158
|
|
|
def from_dict(obj: Any) -> 'Race': |
159
|
|
|
if not isinstance(obj, dict): |
160
|
|
|
return None |
161
|
|
|
name = from_str(obj.get("name")) |
162
|
|
|
slug = from_str(obj.get("slug")) |
163
|
|
|
status = Status.from_dict(obj.get("status")) |
164
|
|
|
url = from_str(obj.get("url")) |
165
|
|
|
data_url = from_str(obj.get("data_url")) |
166
|
|
|
websocket_url = from_union( |
167
|
|
|
[from_str, from_none], obj.get("websocket_url")) |
168
|
|
|
websocket_bot_url = from_union( |
169
|
|
|
[from_str, from_none], obj.get("websocket_bot_url")) |
170
|
|
|
websocket_oauth_url = from_union( |
171
|
|
|
[from_str, from_none], obj.get("websocket_oauth_url")) |
172
|
|
|
category = RaceCategory.from_dict(obj.get("category")) |
173
|
|
|
goal = Goal.from_dict(obj.get("goal")) |
174
|
|
|
info = from_str(obj.get("info")) |
175
|
|
|
entrants_count = from_int(obj.get("entrants_count")) |
176
|
|
|
entrants_count_inactive = from_int(obj.get("entrants_count_inactive")) |
177
|
|
|
entrants = from_union([lambda x: from_list( |
178
|
|
|
Entrant.from_dict, x), from_none], obj.get("entrants")) |
179
|
|
|
opened_at = from_datetime(obj.get("opened_at")) |
180
|
|
|
start_delay = from_union( |
181
|
|
|
[from_timedelta, from_none], obj.get("start_delay")) |
182
|
|
|
started_at = from_union( |
183
|
|
|
[from_datetime, from_none], (obj.get("started_at"))) |
184
|
|
|
ended_at = from_union([from_datetime, from_none], |
185
|
|
|
(obj.get("ended_at"))) |
186
|
|
|
cancelled_at = from_union( |
187
|
|
|
[from_datetime, from_none], (obj.get("cancelled_at"))) |
188
|
|
|
unlisted = from_union([from_bool, from_none], obj.get("unlisted")) |
189
|
|
|
time_limit = from_timedelta(obj.get("time_limit")) |
190
|
|
|
streaming_required = from_union( |
191
|
|
|
[from_bool, from_none], obj.get("streaming_required")) |
192
|
|
|
auto_start = from_union([from_bool, from_none], obj.get("auto_start")) |
193
|
|
|
opened_by = from_union( |
194
|
|
|
[lambda x: User.from_dict(x), from_none], obj.get("opened_by")) |
195
|
|
|
monitors = from_union([lambda x: from_list( |
196
|
|
|
User.from_dict, x), from_none], obj.get("monitors")) |
197
|
|
|
recordable = from_union([from_bool, from_none], obj.get("recordable")) |
198
|
|
|
recorded = from_union([from_bool, from_none], obj.get("recorded")) |
199
|
|
|
recorded_by = from_union( |
200
|
|
|
[lambda x: User.from_dict(x), from_none], obj.get("recorded_by")) |
201
|
|
|
allow_comments = from_union( |
202
|
|
|
[from_bool, from_none], obj.get("allow_comments")) |
203
|
|
|
hide_comments = from_union( |
204
|
|
|
[from_bool, from_none], obj.get("hide_comments")) |
205
|
|
|
allow_midrace_chat = from_union( |
206
|
|
|
[from_bool, from_none], obj.get("allow_midrace_chat")) |
207
|
|
|
allow_non_entrant_chat = from_union( |
208
|
|
|
[from_bool, from_none], obj.get("allow_non_entrant_chat")) |
209
|
|
|
chat_message_delay = from_union( |
210
|
|
|
[from_str, from_none], obj.get("chat_message_delay")) |
211
|
|
|
version = from_union([from_int, from_none], obj.get("version")) |
212
|
|
|
entrants_count_finished = from_union( |
213
|
|
|
[from_int, from_none], obj.get("entrants_count_finished")) |
214
|
|
|
return Race( |
215
|
|
|
name=name, status=status, url=url, data_url=data_url, |
216
|
|
|
websocket_url=websocket_url, websocket_bot_url=websocket_bot_url, |
217
|
|
|
websocket_oauth_url=websocket_oauth_url, category=category, |
218
|
|
|
goal=goal, info=info, entrants_count=entrants_count, |
219
|
|
|
entrants_count_inactive=entrants_count_inactive, |
220
|
|
|
opened_at=opened_at, time_limit=time_limit, entrants=entrants, |
221
|
|
|
version=version, started_at=started_at, ended_at=ended_at, |
222
|
|
|
cancelled_at=cancelled_at, unlisted=unlisted, |
223
|
|
|
streaming_required=streaming_required, auto_start=auto_start, |
224
|
|
|
opened_by=opened_by, monitors=monitors, recordable=recordable, |
225
|
|
|
recorded=recorded, recorded_by=recorded_by, |
226
|
|
|
allow_comments=allow_comments, hide_comments=hide_comments, |
227
|
|
|
allow_midrace_chat=allow_midrace_chat, |
228
|
|
|
allow_non_entrant_chat=allow_non_entrant_chat, |
229
|
|
|
chat_message_delay=chat_message_delay, start_delay=start_delay, |
230
|
|
|
entrants_count_finished=entrants_count_finished, slug=slug |
231
|
|
|
) |
232
|
|
|
|
233
|
|
|
|
234
|
|
|
def race_from_dict(s: Any) -> Race: |
235
|
|
|
return Race.from_dict(s) |
236
|
|
|
|
237
|
|
|
|
238
|
|
|
def races_from_dict(s: Any) -> List[Race]: |
239
|
|
|
assert isinstance(s, dict) |
240
|
|
|
if not s.get("races"): |
241
|
|
|
return [] |
242
|
|
|
return from_union( |
243
|
|
|
[lambda x: from_list(Race.from_dict, x), from_none], s.get("races") |
244
|
|
|
) |
245
|
|
|
|