Bennett/g0v0-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(merge): merge branch 'main' into feat/multiplayer-api

Browse Source
This commit is contained in:
MingxuanGame
2025-08-03 09:50:53 +00:00
parent 41631b839f 9f7ab81213
commit c9fc07b29e
13 changed files with 434 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
app/models/metadata_hub.py
View File

@@ -1,114 +1,85 @@
from __future__ import annotations
from enum import IntEnum
from typing import Any, Literal
from typing import ClassVar, Literal
from app.models.signalr import UserState
from app.models.signalr import SignalRMeta, SignalRUnionMessage, UserState
from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel, 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 _UserActivity(SignalRUnionMessage): ...
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")
union_type: ClassVar[Literal[11]] = 11
class _InGame(_UserActivity):
value: InGameValue = Field(alias="$value")
beatmap_id: int
beatmap_display_title: str
ruleset_id: int
ruleset_playing_verb: str
class InSoloGame(_InGame):
type: Literal["InSoloGame"] = Field(alias="$dtype")
union_type: ClassVar[Literal[12]] = 12
class InMultiplayerGame(_InGame):
type: Literal["InMultiplayerGame"] = Field(alias="$dtype")
union_type: ClassVar[Literal[23]] = 23
class SpectatingMultiplayerGame(_InGame):
type: Literal["SpectatingMultiplayerGame"] = Field(alias="$dtype")
union_type: ClassVar[Literal[24]] = 24
class InPlaylistGame(_InGame):
type: Literal["InPlaylistGame"] = Field(alias="$dtype")
union_type: ClassVar[Literal[31]] = 31
class EditingBeatmapValue(BaseModel):
beatmap_id: int = Field(alias="BeatmapID")
beatmap_display_title: str = Field(alias="BeatmapDisplayTitle")
class PlayingDailyChallenge(_InGame):
union_type: ClassVar[Literal[52]] = 52
class EditingBeatmap(_UserActivity):
type: Literal["EditingBeatmap"] = Field(alias="$dtype")
value: EditingBeatmapValue = Field(alias="$value")
union_type: ClassVar[Literal[41]] = 41
beatmap_id: int
beatmap_display_title: str
class TestingBeatmap(_UserActivity):
type: Literal["TestingBeatmap"] = Field(alias="$dtype")
class TestingBeatmap(EditingBeatmap):
union_type: ClassVar[Literal[43]] = 43
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 ModdingBeatmap(EditingBeatmap):
union_type: ClassVar[Literal[42]] = 42
class WatchingReplay(_UserActivity):
type: Literal["WatchingReplay"] = Field(alias="$dtype")
value: int | None = Field(alias="$value") # Replay ID
union_type: ClassVar[Literal[13]] = 13
score_id: int
player_name: str
beatmap_id: int
beatmap_display_title: str
class SpectatingUser(WatchingReplay):
type: Literal["SpectatingUser"] = Field(alias="$dtype")
union_type: ClassVar[Literal[14]] = 14
class SearchingForLobby(_UserActivity):
type: Literal["SearchingForLobby"] = Field(alias="$dtype")
class InLobbyValue(BaseModel):
room_id: int = Field(alias="RoomID")
room_name: str = Field(alias="RoomName")
union_type: ClassVar[Literal[21]] = 21
class InLobby(_UserActivity):
type: Literal["InLobby"] = "InLobby"
union_type: ClassVar[Literal[22]] = 22
room_id: int
room_name: str
class InDailyChallengeLobby(_UserActivity):
type: Literal["InDailyChallengeLobby"] = Field(alias="$dtype")
union_type: ClassVar[Literal[51]] = 51
UserActivity = (
@@ -128,23 +99,28 @@ UserActivity = (
)
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"),
}
class UserPresence(BaseModel):
activity: UserActivity | None = Field(
default=None, metadata=SignalRMeta(use_upper_case=True)
)
status: OnlineStatus | None = Field(
default=None, metadata=SignalRMeta(use_upper_case=True)
)
@property
def pushable(self) -> bool:
return self.status is not None and self.status != OnlineStatus.OFFLINE
@property
def for_push(self) -> "UserPresence | None":
return UserPresence(
activity=self.activity,
status=self.status,
)
class MetadataClientState(UserPresence, UserState): ...
class OnlineStatus(IntEnum):
OFFLINE = 0 # 隐身

44
app/models/score.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from enum import Enum, IntEnum
from enum import Enum
from typing import Literal, TypedDict
from .mods import API_MODS, APIMod, init_mods
@@ -93,43 +93,6 @@ class HitResult(str, Enum):
)
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 = "friend"
@@ -138,7 +101,6 @@ class LeaderboardType(Enum):
ScoreStatistics = dict[HitResult, int]
ScoreStatisticsInt = dict[HitResultInt, int]
class SoloScoreSubmissionInfo(BaseModel):
@@ -176,8 +138,8 @@ class SoloScoreSubmissionInfo(BaseModel):
class LegacyReplaySoloScoreInfo(TypedDict):
online_id: int
mods: list[APIMod]
statistics: ScoreStatisticsInt
maximum_statistics: ScoreStatisticsInt
statistics: ScoreStatistics
maximum_statistics: ScoreStatistics
client_version: str
rank: Rank
user_id: int

85
app/models/signalr.py
View File

@@ -1,90 +1,23 @@
from __future__ import annotations
import datetime
from enum import Enum
from typing import Any
from dataclasses import dataclass
from typing import ClassVar
from pydantic import (
BaseModel,
BeforeValidator,
ConfigDict,
Field,
TypeAdapter,
model_serializer,
model_validator,
)
def serialize_msgpack(v: Any) -> Any:
typ = v.__class__
if issubclass(typ, BaseModel):
return serialize_to_list(v)
elif issubclass(typ, list):
return TypeAdapter(
typ, config=ConfigDict(arbitrary_types_allowed=True)
).dump_python(v)
elif issubclass(typ, datetime.datetime):
return [v, 0]
elif issubclass(typ, Enum):
list_ = list(typ)
return list_.index(v) if v in list_ else v.value
return v
@dataclass
class SignalRMeta:
member_ignore: bool = False # implement of IgnoreMember (msgpack) attribute
json_ignore: bool = False # implement of JsonIgnore (json) attribute
use_upper_case: bool = False # use upper CamelCase for field names
def serialize_to_list(value: BaseModel) -> list[Any]:
data = []
for field, info in value.__class__.model_fields.items():
data.append(serialize_msgpack(v=getattr(value, field)))
return data
def _by_index(v: Any, class_: type[Enum]):
enum_list = list(class_)
if not isinstance(v, int):
return v
if 0 <= v < len(enum_list):
return enum_list[v]
raise ValueError(
f"Value {v} is out of range for enum "
f"{class_.__name__} with {len(enum_list)} items"
)
def EnumByIndex(enum_class: type[Enum]) -> BeforeValidator:
return BeforeValidator(lambda v: _by_index(v, enum_class))
def msgpack_union(v):
data = v[1]
data.append(v[0])
return data
def msgpack_union_dump(v: BaseModel) -> list[Any]:
_type = getattr(v, "type", None)
if _type is None:
raise ValueError(
f"Model {v.__class__.__name__} does not have a '_type' attribute"
)
return [_type, serialize_to_list(v)]
class MessagePackArrayModel(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
@model_validator(mode="before")
@classmethod
def unpack(cls, v: Any) -> Any:
if isinstance(v, list):
fields = list(cls.model_fields.keys())
if len(v) != len(fields):
raise ValueError(f"Expected list of length {len(fields)}, got {len(v)}")
return dict(zip(fields, v))
return v
@model_serializer
def serialize(self) -> list[Any]:
return serialize_to_list(self)
class SignalRUnionMessage(BaseModel):
union_type: ClassVar[int]
class Transport(BaseModel):

50
app/models/spectator_hub.py
View File

@@ -5,14 +5,14 @@ from enum import IntEnum
from typing import Any
from app.models.beatmap import BeatmapRankStatus
from app.models.mods import APIMod
from .score import (
ScoreStatisticsInt,
ScoreStatistics,
)
from .signalr import MessagePackArrayModel, UserState
from .signalr import SignalRMeta, UserState
from msgpack_lazer_api import APIMod
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import BaseModel, Field, field_validator
class SpectatedUserState(IntEnum):
@@ -24,14 +24,12 @@ class SpectatedUserState(IntEnum):
Quit = 5
class SpectatorState(MessagePackArrayModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
class SpectatorState(BaseModel):
beatmap_id: int | None = None
ruleset_id: int | None = None # 0,1,2,3
mods: list[APIMod] = Field(default_factory=list)
state: SpectatedUserState
maximum_statistics: ScoreStatisticsInt = Field(default_factory=dict)
maximum_statistics: ScoreStatistics = Field(default_factory=dict)
def __eq__(self, other: object) -> bool:
if not isinstance(other, SpectatorState):
@@ -44,22 +42,20 @@ class SpectatorState(MessagePackArrayModel):
)
class ScoreProcessorStatistics(MessagePackArrayModel):
base_score: int
maximum_base_score: int
class ScoreProcessorStatistics(BaseModel):
base_score: float
maximum_base_score: float
accuracy_judgement_count: int
combo_portion: float
bouns_portion: float
bonus_portion: float
class FrameHeader(MessagePackArrayModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
class FrameHeader(BaseModel):
total_score: int
acc: float
accuracy: float
combo: int
max_combo: int
statistics: ScoreStatisticsInt = Field(default_factory=dict)
statistics: ScoreStatistics = Field(default_factory=dict)
score_processor_statistics: ScoreProcessorStatistics
received_time: datetime.datetime
mods: list[APIMod] = Field(default_factory=list)
@@ -87,14 +83,18 @@ class FrameHeader(MessagePackArrayModel):
# SMOKE = 16
class LegacyReplayFrame(MessagePackArrayModel):
class LegacyReplayFrame(BaseModel):
time: float # from ReplayFrame,the parent of LegacyReplayFrame
x: float | None = None
y: float | None = None
mouse_x: float | None = None
mouse_y: float | None = None
button_state: int
header: FrameHeader | None = Field(
default=None, metadata=[SignalRMeta(member_ignore=True)]
)
class FrameDataBundle(MessagePackArrayModel):
class FrameDataBundle(BaseModel):
header: FrameHeader
frames: list[LegacyReplayFrame]
@@ -106,18 +106,16 @@ class APIUser(BaseModel):
class ScoreInfo(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
mods: list[APIMod]
user: APIUser
ruleset: int
maximum_statistics: ScoreStatisticsInt
maximum_statistics: ScoreStatistics
id: int | None = None
total_score: int | None = None
acc: float | None = None
accuracy: float | None = None
max_combo: int | None = None
combo: int | None = None
statistics: ScoreStatisticsInt = Field(default_factory=dict)
statistics: ScoreStatistics = Field(default_factory=dict)
class StoreScore(BaseModel):