107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Player state and registry."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from telnetlib3 import GMCP, MSDP
|
|
|
|
from mudlib.caps import ClientCaps
|
|
from mudlib.entity import Entity
|
|
|
|
if TYPE_CHECKING:
|
|
from mudlib.editor import Editor
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.if_session import IFSession
|
|
|
|
|
|
@dataclass(eq=False)
|
|
class Player(Entity):
|
|
"""Represents a connected player."""
|
|
|
|
writer: Any = None # telnetlib3 TelnetWriter for sending output
|
|
reader: Any = None # telnetlib3 TelnetReader for reading input
|
|
flying: bool = False
|
|
mode_stack: list[str] = field(default_factory=lambda: ["normal"])
|
|
caps: ClientCaps = field(default_factory=lambda: ClientCaps(ansi=True))
|
|
editor: Editor | None = None
|
|
if_session: IFSession | EmbeddedIFSession | None = None
|
|
paint_mode: bool = False
|
|
painting: bool = False
|
|
paint_brush: str = "."
|
|
prompt_template: str | None = None
|
|
aliases: dict[str, str] = field(default_factory=dict)
|
|
_last_msdp: dict = field(default_factory=dict, repr=False)
|
|
_power_task: asyncio.Task | None = None
|
|
kills: int = 0
|
|
deaths: int = 0
|
|
mob_kills: dict[str, int] = field(default_factory=dict)
|
|
play_time_seconds: float = 0.0
|
|
unlocked_moves: set[str] = field(default_factory=set)
|
|
session_start: float = 0.0
|
|
is_admin: bool = False
|
|
|
|
@property
|
|
def mode(self) -> str:
|
|
"""Current mode is the top of the stack."""
|
|
return self.mode_stack[-1]
|
|
|
|
@property
|
|
def color_depth(self) -> str | None:
|
|
"""Best available color mode: truecolor, 256, 16, or None if no ANSI."""
|
|
if not self.caps.ansi:
|
|
return None
|
|
return self.caps.color_depth
|
|
|
|
async def send(self, message: str) -> None:
|
|
"""Send a message to the player via their telnet writer."""
|
|
self.writer.write(message)
|
|
await self.writer.drain()
|
|
|
|
def send_gmcp(self, package: str, data: Any = None) -> None:
|
|
"""Send a GMCP message to the client (no-op if unsupported)."""
|
|
if self.writer is not None:
|
|
self.writer.send_gmcp(package, data)
|
|
|
|
def send_msdp(self, variables: dict) -> None:
|
|
"""Send MSDP variables to the client (no-op if unsupported)."""
|
|
if self.writer is not None:
|
|
self.writer.send_msdp(variables)
|
|
|
|
@property
|
|
def gmcp_enabled(self) -> bool:
|
|
"""Whether this client has GMCP negotiated."""
|
|
if self.writer is None:
|
|
return False
|
|
return bool(
|
|
self.writer.local_option.enabled(GMCP)
|
|
or self.writer.remote_option.enabled(GMCP)
|
|
)
|
|
|
|
@property
|
|
def msdp_enabled(self) -> bool:
|
|
"""Whether this client has MSDP negotiated."""
|
|
if self.writer is None:
|
|
return False
|
|
return bool(
|
|
self.writer.local_option.enabled(MSDP)
|
|
or self.writer.remote_option.enabled(MSDP)
|
|
)
|
|
|
|
|
|
def accumulate_play_time(player: Player) -> None:
|
|
"""Accumulate play time since session start and reset timer.
|
|
|
|
Args:
|
|
player: The player to update
|
|
"""
|
|
if player.session_start > 0:
|
|
player.play_time_seconds += time.monotonic() - player.session_start
|
|
player.session_start = time.monotonic()
|
|
|
|
|
|
# Global registry of connected players
|
|
players: dict[str, Player] = {}
|