Merge branch 'main' into feat/multiplayer-api
This commit is contained in:
jimmy-sketch
152
app/models/metadata_hub.py
Normal file
152
app/models/metadata_hub.py
Normal file
@@ -0,0 +1,152 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import IntEnum
|
||||
from typing import Any, Literal
|
||||
|
||||
from app.models.signalr import UserState
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
class _UserActivity(BaseModel):
|
||||
model_config = ConfigDict(serialize_by_alias=True)
|
||||
type: Literal[
|
||||
"ChoosingBeatmap",
|
||||
"InSoloGame",
|
||||
"WatchingReplay",
|
||||
"SpectatingUser",
|
||||
"SearchingForLobby",
|
||||
"InLobby",
|
||||
"InMultiplayerGame",
|
||||
"SpectatingMultiplayerGame",
|
||||
"InPlaylistGame",
|
||||
"EditingBeatmap",
|
||||
"ModdingBeatmap",
|
||||
"TestingBeatmap",
|
||||
"InDailyChallengeLobby",
|
||||
"PlayingDailyChallenge",
|
||||
] = Field(alias="$dtype")
|
||||
value: Any | None = Field(alias="$value")
|
||||
|
||||
|
||||
class ChoosingBeatmap(_UserActivity):
|
||||
type: Literal["ChoosingBeatmap"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class InGameValue(BaseModel):
|
||||
beatmap_id: int = Field(alias="BeatmapID")
|
||||
beatmap_display_title: str = Field(alias="BeatmapDisplayTitle")
|
||||
ruleset_id: int = Field(alias="RulesetID")
|
||||
ruleset_playing_verb: str = Field(alias="RulesetPlayingVerb")
|
||||
|
||||
|
||||
class _InGame(_UserActivity):
|
||||
value: InGameValue = Field(alias="$value")
|
||||
|
||||
|
||||
class InSoloGame(_InGame):
|
||||
type: Literal["InSoloGame"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class InMultiplayerGame(_InGame):
|
||||
type: Literal["InMultiplayerGame"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class SpectatingMultiplayerGame(_InGame):
|
||||
type: Literal["SpectatingMultiplayerGame"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class InPlaylistGame(_InGame):
|
||||
type: Literal["InPlaylistGame"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class EditingBeatmapValue(BaseModel):
|
||||
beatmap_id: int = Field(alias="BeatmapID")
|
||||
beatmap_display_title: str = Field(alias="BeatmapDisplayTitle")
|
||||
|
||||
|
||||
class EditingBeatmap(_UserActivity):
|
||||
type: Literal["EditingBeatmap"] = Field(alias="$dtype")
|
||||
value: EditingBeatmapValue = Field(alias="$value")
|
||||
|
||||
|
||||
class TestingBeatmap(_UserActivity):
|
||||
type: Literal["TestingBeatmap"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class ModdingBeatmap(_UserActivity):
|
||||
type: Literal["ModdingBeatmap"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class WatchingReplayValue(BaseModel):
|
||||
score_id: int = Field(alias="ScoreID")
|
||||
player_name: str = Field(alias="PlayerName")
|
||||
beatmap_id: int = Field(alias="BeatmapID")
|
||||
beatmap_display_title: str = Field(alias="BeatmapDisplayTitle")
|
||||
|
||||
|
||||
class WatchingReplay(_UserActivity):
|
||||
type: Literal["WatchingReplay"] = Field(alias="$dtype")
|
||||
value: int | None = Field(alias="$value") # Replay ID
|
||||
|
||||
|
||||
class SpectatingUser(WatchingReplay):
|
||||
type: Literal["SpectatingUser"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class SearchingForLobby(_UserActivity):
|
||||
type: Literal["SearchingForLobby"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
class InLobbyValue(BaseModel):
|
||||
room_id: int = Field(alias="RoomID")
|
||||
room_name: str = Field(alias="RoomName")
|
||||
|
||||
|
||||
class InLobby(_UserActivity):
|
||||
type: Literal["InLobby"] = "InLobby"
|
||||
|
||||
|
||||
class InDailyChallengeLobby(_UserActivity):
|
||||
type: Literal["InDailyChallengeLobby"] = Field(alias="$dtype")
|
||||
|
||||
|
||||
UserActivity = (
|
||||
ChoosingBeatmap
|
||||
| InSoloGame
|
||||
| WatchingReplay
|
||||
| SpectatingUser
|
||||
| SearchingForLobby
|
||||
| InLobby
|
||||
| InMultiplayerGame
|
||||
| SpectatingMultiplayerGame
|
||||
| InPlaylistGame
|
||||
| EditingBeatmap
|
||||
| ModdingBeatmap
|
||||
| TestingBeatmap
|
||||
| InDailyChallengeLobby
|
||||
)
|
||||
|
||||
|
||||
class MetadataClientState(UserState):
|
||||
user_activity: UserActivity | None = None
|
||||
status: OnlineStatus | None = None
|
||||
|
||||
def to_dict(self) -> dict[str, Any] | None:
|
||||
if self.status is None or self.status == OnlineStatus.OFFLINE:
|
||||
return None
|
||||
dumped = self.model_dump(by_alias=True, exclude_none=True)
|
||||
return {
|
||||
"Activity": dumped.get("user_activity"),
|
||||
"Status": dumped.get("status"),
|
||||
}
|
||||
|
||||
@property
|
||||
def pushable(self) -> bool:
|
||||
return self.status is not None and self.status != OnlineStatus.OFFLINE
|
||||
|
||||
|
||||
class OnlineStatus(IntEnum):
|
||||
OFFLINE = 0 # 隐身
|
||||
DO_NOT_DISTURB = 1
|
||||
ONLINE = 2
|
||||
@@ -1,47 +1,91 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TypedDict
|
||||
import json
|
||||
from typing import Literal, NotRequired, TypedDict
|
||||
|
||||
from app.path import STATIC_DIR
|
||||
|
||||
|
||||
class APIMod(TypedDict):
|
||||
acronym: str
|
||||
settings: dict[str, bool | float | str]
|
||||
settings: NotRequired[dict[str, bool | float | str]]
|
||||
|
||||
|
||||
# https://github.com/ppy/osu-api/wiki#mods
|
||||
LEGACY_MOD_TO_API_MOD = {
|
||||
(1 << 0): APIMod(acronym="NF", settings={}), # No Fail
|
||||
(1 << 1): APIMod(acronym="EZ", settings={}),
|
||||
(1 << 2): APIMod(acronym="TD", settings={}), # Touch Device
|
||||
(1 << 3): APIMod(acronym="HD", settings={}), # Hidden
|
||||
(1 << 4): APIMod(acronym="HR", settings={}), # Hard Rock
|
||||
(1 << 5): APIMod(acronym="SD", settings={}), # Sudden Death
|
||||
(1 << 6): APIMod(acronym="DT", settings={}), # Double Time
|
||||
(1 << 7): APIMod(acronym="RX", settings={}), # Relax
|
||||
(1 << 8): APIMod(acronym="HT", settings={}), # Half Time
|
||||
(1 << 9): APIMod(acronym="NC", settings={}), # Nightcore
|
||||
(1 << 10): APIMod(acronym="FL", settings={}), # Flashlight
|
||||
(1 << 11): APIMod(acronym="AT", settings={}), # Auto Play
|
||||
(1 << 12): APIMod(acronym="SO", settings={}), # Spun Out
|
||||
(1 << 13): APIMod(acronym="AP", settings={}), # Autopilot
|
||||
(1 << 14): APIMod(acronym="PF", settings={}), # Perfect
|
||||
(1 << 15): APIMod(acronym="4K", settings={}), # 4K
|
||||
(1 << 16): APIMod(acronym="5K", settings={}), # 5K
|
||||
(1 << 17): APIMod(acronym="6K", settings={}), # 6K
|
||||
(1 << 18): APIMod(acronym="7K", settings={}), # 7K
|
||||
(1 << 19): APIMod(acronym="8K", settings={}), # 8K
|
||||
(1 << 20): APIMod(acronym="FI", settings={}), # Fade In
|
||||
(1 << 21): APIMod(acronym="RD", settings={}), # Random
|
||||
(1 << 22): APIMod(acronym="CN", settings={}), # Cinema
|
||||
(1 << 23): APIMod(acronym="TP", settings={}), # Target Practice
|
||||
(1 << 24): APIMod(acronym="9K", settings={}), # 9K
|
||||
(1 << 25): APIMod(acronym="CO", settings={}), # Key Co-op
|
||||
(1 << 26): APIMod(acronym="1K", settings={}), # 1K
|
||||
(1 << 27): APIMod(acronym="2K", settings={}), # 2K
|
||||
(1 << 28): APIMod(acronym="3K", settings={}), # 3K
|
||||
(1 << 29): APIMod(acronym="SV2", settings={}), # Score V2
|
||||
(1 << 30): APIMod(acronym="MR", settings={}), # Mirror
|
||||
API_MOD_TO_LEGACY: dict[str, int] = {
|
||||
"NF": 1 << 0, # No Fail
|
||||
"EZ": 1 << 1, # Easy
|
||||
"TD": 1 << 2, # Touch Device
|
||||
"HD": 1 << 3, # Hidden
|
||||
"HR": 1 << 4, # Hard Rock
|
||||
"SD": 1 << 5, # Sudden Death
|
||||
"DT": 1 << 6, # Double Time
|
||||
"RX": 1 << 7, # Relax
|
||||
"HT": 1 << 8, # Half Time
|
||||
"NC": 1 << 9, # Nightcore
|
||||
"FL": 1 << 10, # Flashlight
|
||||
"AT": 1 << 11, # Autoplay
|
||||
"SO": 1 << 12, # Spun Out
|
||||
"AP": 1 << 13, # Auto Pilot
|
||||
"PF": 1 << 14, # Perfect
|
||||
"4K": 1 << 15, # 4K
|
||||
"5K": 1 << 16, # 5K
|
||||
"6K": 1 << 17, # 6K
|
||||
"7K": 1 << 18, # 7K
|
||||
"8K": 1 << 19, # 8K
|
||||
"FI": 1 << 20, # Fade In
|
||||
"RD": 1 << 21, # Random
|
||||
"CN": 1 << 22, # Cinema
|
||||
"TP": 1 << 23, # Target Practice
|
||||
"9K": 1 << 24, # 9K
|
||||
"CO": 1 << 25, # Key Co-op
|
||||
"1K": 1 << 26, # 1K
|
||||
"3K": 1 << 27, # 3K
|
||||
"2K": 1 << 28, # 2K
|
||||
"SV2": 1 << 29, # ScoreV2
|
||||
"MR": 1 << 30, # Mirror
|
||||
}
|
||||
LEGACY_MOD_TO_API_MOD = {}
|
||||
for k, v in API_MOD_TO_LEGACY.items():
|
||||
LEGACY_MOD_TO_API_MOD[v] = APIMod(acronym=k, settings={})
|
||||
API_MOD_TO_LEGACY["NC"] |= API_MOD_TO_LEGACY["DT"]
|
||||
API_MOD_TO_LEGACY["PF"] |= API_MOD_TO_LEGACY["SD"]
|
||||
|
||||
|
||||
# see static/mods.json
|
||||
class Settings(TypedDict):
|
||||
Name: str
|
||||
Type: str
|
||||
Label: str
|
||||
Description: str
|
||||
|
||||
|
||||
class Mod(TypedDict):
|
||||
Acronym: str
|
||||
Name: str
|
||||
Description: str
|
||||
Type: str
|
||||
Settings: list[Settings]
|
||||
IncompatibleMods: list[str]
|
||||
RequiresConfiguration: bool
|
||||
UserPlayable: bool
|
||||
ValidForMultiplayer: bool
|
||||
ValidForFreestyleAsRequiredMod: bool
|
||||
ValidForMultiplayerAsFreeMod: bool
|
||||
AlwaysValidForSubmission: bool
|
||||
|
||||
|
||||
API_MODS: dict[Literal[0, 1, 2, 3], dict[str, Mod]] = {}
|
||||
|
||||
|
||||
def init_mods():
|
||||
mods_file = STATIC_DIR / "mods.json"
|
||||
raw_mods = json.loads(mods_file.read_text())
|
||||
for ruleset in raw_mods:
|
||||
ruleset_mods = {}
|
||||
for mod in ruleset["Mods"]:
|
||||
ruleset_mods[mod["Acronym"]] = mod
|
||||
API_MODS[ruleset["RulesetID"]] = ruleset_mods
|
||||
|
||||
|
||||
def int_to_mods(mods: int) -> list[APIMod]:
|
||||
@@ -54,3 +98,10 @@ def int_to_mods(mods: int) -> list[APIMod]:
|
||||
if mods & (1 << 9):
|
||||
mod_list.remove(LEGACY_MOD_TO_API_MOD[(1 << 6)])
|
||||
return mod_list
|
||||
|
||||
|
||||
def mods_to_int(mods: list[APIMod]) -> int:
|
||||
sum_ = 0
|
||||
for mod in mods:
|
||||
sum_ |= API_MOD_TO_LEGACY.get(mod["acronym"], 0)
|
||||
return sum_
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
# OAuth 相关模型
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
@@ -34,3 +35,22 @@ class OAuthErrorResponse(BaseModel):
|
||||
error_description: str
|
||||
hint: str
|
||||
message: str
|
||||
|
||||
|
||||
class RegistrationErrorResponse(BaseModel):
|
||||
"""注册错误响应模型"""
|
||||
form_error: dict
|
||||
|
||||
|
||||
class UserRegistrationErrors(BaseModel):
|
||||
"""用户注册错误模型"""
|
||||
username: List[str] = []
|
||||
user_email: List[str] = []
|
||||
password: List[str] = []
|
||||
|
||||
|
||||
class RegistrationRequestErrors(BaseModel):
|
||||
"""注册请求错误模型"""
|
||||
message: str | None = None
|
||||
redirect: str | None = None
|
||||
user: UserRegistrationErrors | None = None
|
||||
|
||||
@@ -1,7 +1,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import Enum, IntEnum
|
||||
from typing import Literal, TypedDict
|
||||
|
||||
from .mods import API_MODS, APIMod, init_mods
|
||||
|
||||
from pydantic import BaseModel, Field, ValidationInfo, field_validator
|
||||
import rosu_pp_py as rosu
|
||||
|
||||
|
||||
@@ -30,40 +34,141 @@ INT_TO_MODE = {v: k for k, v in MODE_TO_INT.items()}
|
||||
|
||||
|
||||
class Rank(str, Enum):
|
||||
X = "ss"
|
||||
XH = "ssh"
|
||||
S = "s"
|
||||
SH = "sh"
|
||||
A = "a"
|
||||
B = "b"
|
||||
C = "c"
|
||||
D = "d"
|
||||
F = "f"
|
||||
X = "X"
|
||||
XH = "XH"
|
||||
S = "S"
|
||||
SH = "SH"
|
||||
A = "A"
|
||||
B = "B"
|
||||
C = "C"
|
||||
D = "D"
|
||||
F = "F"
|
||||
|
||||
|
||||
# https://github.com/ppy/osu/blob/master/osu.Game/Rulesets/Scoring/HitResult.cs
|
||||
class HitResult(IntEnum):
|
||||
PERFECT = 0 # [Order(0)]
|
||||
GREAT = 1 # [Order(1)]
|
||||
GOOD = 2 # [Order(2)]
|
||||
OK = 3 # [Order(3)]
|
||||
MEH = 4 # [Order(4)]
|
||||
MISS = 5 # [Order(5)]
|
||||
class HitResult(str, Enum):
|
||||
PERFECT = "perfect" # [Order(0)]
|
||||
GREAT = "great" # [Order(1)]
|
||||
GOOD = "good" # [Order(2)]
|
||||
OK = "ok" # [Order(3)]
|
||||
MEH = "meh" # [Order(4)]
|
||||
MISS = "miss" # [Order(5)]
|
||||
|
||||
LARGE_TICK_HIT = 6 # [Order(6)]
|
||||
SMALL_TICK_HIT = 7 # [Order(7)]
|
||||
SLIDER_TAIL_HIT = 8 # [Order(8)]
|
||||
LARGE_TICK_HIT = "large_tick_hit" # [Order(6)]
|
||||
SMALL_TICK_HIT = "small_tick_hit" # [Order(7)]
|
||||
SLIDER_TAIL_HIT = "slider_tail_hit" # [Order(8)]
|
||||
|
||||
LARGE_BONUS = 9 # [Order(9)]
|
||||
SMALL_BONUS = 10 # [Order(10)]
|
||||
LARGE_BONUS = "large_bonus" # [Order(9)]
|
||||
SMALL_BONUS = "small_bonus" # [Order(10)]
|
||||
|
||||
LARGE_TICK_MISS = 11 # [Order(11)]
|
||||
SMALL_TICK_MISS = 12 # [Order(12)]
|
||||
LARGE_TICK_MISS = "large_tick_miss" # [Order(11)]
|
||||
SMALL_TICK_MISS = "small_tick_miss" # [Order(12)]
|
||||
|
||||
IGNORE_HIT = 13 # [Order(13)]
|
||||
IGNORE_MISS = 14 # [Order(14)]
|
||||
IGNORE_HIT = "ignore_hit" # [Order(13)]
|
||||
IGNORE_MISS = "ignore_miss" # [Order(14)]
|
||||
|
||||
NONE = 15 # [Order(15)]
|
||||
COMBO_BREAK = 16 # [Order(16)]
|
||||
NONE = "none" # [Order(15)]
|
||||
COMBO_BREAK = "combo_break" # [Order(16)]
|
||||
|
||||
LEGACY_COMBO_INCREASE = 99 # [Order(99)] @deprecated
|
||||
LEGACY_COMBO_INCREASE = "legacy_combo_increase" # [Order(99)] @deprecated
|
||||
|
||||
def is_hit(self) -> bool:
|
||||
return self not in (
|
||||
HitResult.NONE,
|
||||
HitResult.IGNORE_MISS,
|
||||
HitResult.COMBO_BREAK,
|
||||
HitResult.LARGE_TICK_MISS,
|
||||
HitResult.SMALL_TICK_MISS,
|
||||
HitResult.MISS,
|
||||
)
|
||||
|
||||
|
||||
class HitResultInt(IntEnum):
|
||||
PERFECT = 0
|
||||
GREAT = 1
|
||||
GOOD = 2
|
||||
OK = 3
|
||||
MEH = 4
|
||||
MISS = 5
|
||||
|
||||
LARGE_TICK_HIT = 6
|
||||
SMALL_TICK_HIT = 7
|
||||
SLIDER_TAIL_HIT = 8
|
||||
|
||||
LARGE_BONUS = 9
|
||||
SMALL_BONUS = 10
|
||||
|
||||
LARGE_TICK_MISS = 11
|
||||
SMALL_TICK_MISS = 12
|
||||
|
||||
IGNORE_HIT = 13
|
||||
IGNORE_MISS = 14
|
||||
|
||||
NONE = 15
|
||||
COMBO_BREAK = 16
|
||||
|
||||
LEGACY_COMBO_INCREASE = 99
|
||||
|
||||
def is_hit(self) -> bool:
|
||||
return self not in (
|
||||
HitResultInt.NONE,
|
||||
HitResultInt.IGNORE_MISS,
|
||||
HitResultInt.COMBO_BREAK,
|
||||
HitResultInt.LARGE_TICK_MISS,
|
||||
HitResultInt.SMALL_TICK_MISS,
|
||||
HitResultInt.MISS,
|
||||
)
|
||||
|
||||
|
||||
class LeaderboardType(Enum):
|
||||
GLOBAL = "global"
|
||||
FRIENDS = "friends"
|
||||
COUNTRY = "country"
|
||||
TEAM = "team"
|
||||
|
||||
|
||||
ScoreStatistics = dict[HitResult, int]
|
||||
ScoreStatisticsInt = dict[HitResultInt, int]
|
||||
|
||||
|
||||
class SoloScoreSubmissionInfo(BaseModel):
|
||||
rank: Rank
|
||||
total_score: int = Field(ge=0, le=2**31 - 1)
|
||||
total_score_without_mods: int = Field(ge=0, le=2**31 - 1)
|
||||
accuracy: float = Field(ge=0, le=1)
|
||||
pp: float = Field(default=0, ge=0, le=2**31 - 1)
|
||||
max_combo: int = 0
|
||||
ruleset_id: Literal[0, 1, 2, 3]
|
||||
passed: bool = False
|
||||
mods: list[APIMod] = Field(default_factory=list)
|
||||
statistics: ScoreStatistics = Field(default_factory=dict)
|
||||
maximum_statistics: ScoreStatistics = Field(default_factory=dict)
|
||||
|
||||
@field_validator("mods", mode="after")
|
||||
@classmethod
|
||||
def validate_mods(cls, mods: list[APIMod], info: ValidationInfo):
|
||||
if not API_MODS:
|
||||
init_mods()
|
||||
incompatible_mods = set()
|
||||
# check incompatible mods
|
||||
for mod in mods:
|
||||
if mod["acronym"] in incompatible_mods:
|
||||
raise ValueError(
|
||||
f"Mod {mod['acronym']} is incompatible with other mods"
|
||||
)
|
||||
setting_mods = API_MODS[info.data["ruleset_id"]].get(mod["acronym"])
|
||||
if not setting_mods:
|
||||
raise ValueError(f"Invalid mod: {mod['acronym']}")
|
||||
incompatible_mods.update(setting_mods["IncompatibleMods"])
|
||||
return mods
|
||||
|
||||
|
||||
class LegacyReplaySoloScoreInfo(TypedDict):
|
||||
online_id: int
|
||||
mods: list[APIMod]
|
||||
statistics: ScoreStatisticsInt
|
||||
maximum_statistics: ScoreStatisticsInt
|
||||
client_version: str
|
||||
rank: Rank
|
||||
user_id: int
|
||||
total_score_without_mods: int
|
||||
|
||||
@@ -1,11 +1,42 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
import datetime
|
||||
from typing import Any, get_origin
|
||||
|
||||
from pydantic import BaseModel, Field, model_validator
|
||||
import msgpack
|
||||
from pydantic import (
|
||||
BaseModel,
|
||||
ConfigDict,
|
||||
Field,
|
||||
TypeAdapter,
|
||||
model_serializer,
|
||||
model_validator,
|
||||
)
|
||||
|
||||
|
||||
def serialize_to_list(value: BaseModel) -> list[Any]:
|
||||
data = []
|
||||
for field, info in value.__class__.model_fields.items():
|
||||
v = getattr(value, field)
|
||||
anno = get_origin(info.annotation)
|
||||
if anno and issubclass(anno, BaseModel):
|
||||
data.append(serialize_to_list(v))
|
||||
elif anno and issubclass(anno, list):
|
||||
data.append(
|
||||
TypeAdapter(
|
||||
info.annotation,
|
||||
).dump_python(v)
|
||||
)
|
||||
elif isinstance(v, datetime.datetime):
|
||||
data.append([msgpack.ext.Timestamp.from_datetime(v), 0])
|
||||
else:
|
||||
data.append(v)
|
||||
return data
|
||||
|
||||
|
||||
class MessagePackArrayModel(BaseModel):
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def unpack(cls, v: Any) -> Any:
|
||||
@@ -16,11 +47,15 @@ class MessagePackArrayModel(BaseModel):
|
||||
return dict(zip(fields, v))
|
||||
return v
|
||||
|
||||
@model_serializer
|
||||
def serialize(self) -> list[Any]:
|
||||
return serialize_to_list(self)
|
||||
|
||||
|
||||
class Transport(BaseModel):
|
||||
transport: str
|
||||
transfer_formats: list[str] = Field(
|
||||
default_factory=lambda: ["Binary"], alias="transferFormats"
|
||||
default_factory=lambda: ["Binary", "Text"], alias="transferFormats"
|
||||
)
|
||||
|
||||
|
||||
@@ -29,3 +64,8 @@ class NegotiateResponse(BaseModel):
|
||||
connectionToken: str
|
||||
negotiateVersion: int = 1
|
||||
availableTransports: list[Transport]
|
||||
|
||||
|
||||
class UserState(BaseModel):
|
||||
connection_id: str
|
||||
connection_token: str
|
||||
|
||||
@@ -4,18 +4,22 @@ import datetime
|
||||
from enum import IntEnum
|
||||
from typing import Any
|
||||
|
||||
from app.models.beatmap import BeatmapRankStatus
|
||||
|
||||
from .score import (
|
||||
HitResult,
|
||||
ScoreStatisticsInt,
|
||||
)
|
||||
from .signalr import MessagePackArrayModel
|
||||
from .signalr import MessagePackArrayModel, UserState
|
||||
|
||||
import msgpack
|
||||
from pydantic import Field, field_validator
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
|
||||
class APIMod(MessagePackArrayModel):
|
||||
acronym: str
|
||||
settings: dict[str, Any] = Field(default_factory=dict)
|
||||
settings: dict[str, Any] | list = Field(
|
||||
default_factory=dict
|
||||
) # FIXME: with settings
|
||||
|
||||
|
||||
class SpectatedUserState(IntEnum):
|
||||
@@ -32,7 +36,7 @@ class SpectatorState(MessagePackArrayModel):
|
||||
ruleset_id: int | None = None # 0,1,2,3
|
||||
mods: list[APIMod] = Field(default_factory=list)
|
||||
state: SpectatedUserState
|
||||
maximum_statistics: dict[HitResult, int] = Field(default_factory=dict)
|
||||
maximum_statistics: ScoreStatisticsInt = Field(default_factory=dict)
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if not isinstance(other, SpectatorState):
|
||||
@@ -58,7 +62,7 @@ class FrameHeader(MessagePackArrayModel):
|
||||
acc: float
|
||||
combo: int
|
||||
max_combo: int
|
||||
statistics: dict[HitResult, int] = Field(default_factory=dict)
|
||||
statistics: ScoreStatisticsInt = Field(default_factory=dict)
|
||||
score_processor_statistics: ScoreProcessorStatistics
|
||||
received_time: datetime.datetime
|
||||
mods: list[APIMod] = Field(default_factory=list)
|
||||
@@ -79,22 +83,56 @@ class FrameHeader(MessagePackArrayModel):
|
||||
raise ValueError(f"Cannot convert {type(v)} to datetime")
|
||||
|
||||
|
||||
class ReplayButtonState(IntEnum):
|
||||
NONE = 0
|
||||
LEFT1 = 1
|
||||
RIGHT1 = 2
|
||||
LEFT2 = 4
|
||||
RIGHT2 = 8
|
||||
SMOKE = 16
|
||||
# class ReplayButtonState(IntEnum):
|
||||
# NONE = 0
|
||||
# LEFT1 = 1
|
||||
# RIGHT1 = 2
|
||||
# LEFT2 = 4
|
||||
# RIGHT2 = 8
|
||||
# SMOKE = 16
|
||||
|
||||
|
||||
class LegacyReplayFrame(MessagePackArrayModel):
|
||||
time: int # from ReplayFrame,the parent of LegacyReplayFrame
|
||||
time: float # from ReplayFrame,the parent of LegacyReplayFrame
|
||||
x: float | None = None
|
||||
y: float | None = None
|
||||
button_state: ReplayButtonState
|
||||
button_state: int
|
||||
|
||||
|
||||
class FrameDataBundle(MessagePackArrayModel):
|
||||
header: FrameHeader
|
||||
frames: list[LegacyReplayFrame]
|
||||
|
||||
|
||||
# Use for server
|
||||
class APIUser(BaseModel):
|
||||
id: int
|
||||
name: str
|
||||
|
||||
|
||||
class ScoreInfo(BaseModel):
|
||||
mods: list[APIMod]
|
||||
user: APIUser
|
||||
ruleset: int
|
||||
maximum_statistics: ScoreStatisticsInt
|
||||
id: int | None = None
|
||||
total_score: int | None = None
|
||||
acc: float | None = None
|
||||
max_combo: int | None = None
|
||||
combo: int | None = None
|
||||
statistics: ScoreStatisticsInt = Field(default_factory=dict)
|
||||
|
||||
|
||||
class StoreScore(BaseModel):
|
||||
score_info: ScoreInfo
|
||||
replay_frames: list[LegacyReplayFrame] = Field(default_factory=list)
|
||||
|
||||
|
||||
class StoreClientState(UserState):
|
||||
state: SpectatorState | None = None
|
||||
beatmap_status: BeatmapRankStatus | None = None
|
||||
checksum: str | None = None
|
||||
ruleset_id: int | None = None
|
||||
score_token: int | None = None
|
||||
watched_user: set[int] = Field(default_factory=set)
|
||||
score: StoreScore | None = None
|
||||
|
||||
Reference in New Issue
Block a user