126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
from datetime import UTC, datetime
|
|
|
|
from app.models.multiplayer_hub import ServerMultiplayerRoom
|
|
from app.models.room import (
|
|
MatchType,
|
|
QueueMode,
|
|
RoomCategory,
|
|
RoomDifficultyRange,
|
|
RoomPlaylistItemStats,
|
|
RoomStatus,
|
|
)
|
|
|
|
from .lazer_user import User, UserResp
|
|
from .playlists import Playlist, PlaylistResp
|
|
|
|
from sqlmodel import (
|
|
BigInteger,
|
|
Column,
|
|
DateTime,
|
|
Field,
|
|
ForeignKey,
|
|
Relationship,
|
|
SQLModel,
|
|
)
|
|
|
|
|
|
class RoomBase(SQLModel):
|
|
name: str = Field(index=True)
|
|
category: RoomCategory = Field(default=RoomCategory.NORMAL, index=True)
|
|
duration: int | None = Field(default=None) # minutes
|
|
starts_at: datetime | None = Field(
|
|
sa_column=Column(
|
|
DateTime(timezone=True),
|
|
),
|
|
default=datetime.now(UTC),
|
|
)
|
|
ended_at: datetime | None = Field(
|
|
sa_column=Column(
|
|
DateTime(timezone=True),
|
|
),
|
|
default=None,
|
|
)
|
|
participant_count: int = Field(default=0)
|
|
max_attempts: int | None = Field(default=None) # playlists
|
|
type: MatchType
|
|
queue_mode: QueueMode
|
|
auto_skip: bool
|
|
auto_start_duration: int
|
|
status: RoomStatus
|
|
# TODO: channel_id
|
|
# recent_participants: list[User]
|
|
|
|
|
|
class Room(RoomBase, table=True):
|
|
__tablename__ = "rooms" # pyright: ignore[reportAssignmentType]
|
|
id: int = Field(default=None, primary_key=True, index=True)
|
|
host_id: int = Field(
|
|
sa_column=Column(BigInteger, ForeignKey("lazer_users.id"), index=True)
|
|
)
|
|
|
|
host: User = Relationship()
|
|
playlist: list[Playlist] = Relationship(
|
|
sa_relationship_kwargs={
|
|
"lazy": "joined",
|
|
"cascade": "all, delete-orphan",
|
|
"overlaps": "room",
|
|
}
|
|
)
|
|
|
|
|
|
class RoomResp(RoomBase):
|
|
id: int
|
|
password: str | None = None
|
|
host: UserResp | None = None
|
|
playlist: list[PlaylistResp] = []
|
|
playlist_item_stats: RoomPlaylistItemStats | None = None
|
|
difficulty_range: RoomDifficultyRange | None = None
|
|
current_playlist_item: PlaylistResp | None = None
|
|
|
|
@classmethod
|
|
async def from_db(cls, room: Room) -> "RoomResp":
|
|
resp = cls.model_validate(room.model_dump())
|
|
|
|
stats = RoomPlaylistItemStats(count_active=0, count_total=0)
|
|
difficulty_range = RoomDifficultyRange(
|
|
min=0,
|
|
max=0,
|
|
)
|
|
rulesets = set()
|
|
for playlist in room.playlist:
|
|
stats.count_total += 1
|
|
if not playlist.expired:
|
|
stats.count_active += 1
|
|
rulesets.add(playlist.ruleset_id)
|
|
difficulty_range.min = min(
|
|
difficulty_range.min, playlist.beatmap.difficulty_rating
|
|
)
|
|
difficulty_range.max = max(
|
|
difficulty_range.max, playlist.beatmap.difficulty_rating
|
|
)
|
|
resp.playlist.append(await PlaylistResp.from_db(playlist, ["beatmap"]))
|
|
stats.ruleset_ids = list(rulesets)
|
|
resp.playlist_item_stats = stats
|
|
resp.difficulty_range = difficulty_range
|
|
resp.current_playlist_item = resp.playlist[-1] if resp.playlist else None
|
|
|
|
return resp
|
|
|
|
@classmethod
|
|
async def from_hub(cls, server_room: ServerMultiplayerRoom) -> "RoomResp":
|
|
room = server_room.room
|
|
resp = cls(
|
|
id=room.room_id,
|
|
name=room.settings.name,
|
|
type=room.settings.match_type,
|
|
queue_mode=room.settings.queue_mode,
|
|
auto_skip=room.settings.auto_skip,
|
|
auto_start_duration=int(room.settings.auto_start_duration.total_seconds()),
|
|
status=server_room.status,
|
|
category=server_room.category,
|
|
# duration = room.settings.duration,
|
|
starts_at=server_room.start_at,
|
|
participant_count=len(room.users),
|
|
)
|
|
return resp
|