vladimir/mind-reader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

wip: без комментариев

Browse Source
This commit is contained in:
vladimir
2026-08-23 01:49:02 +03:00
parent 5c7966755f
commit 9a4c98fb76
15 changed files with 491 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mind_reader/domain/__init__.py
+14
View File
@@ -4,7 +4,14 @@ from .session import (
ArrowDirection,
CardData,
CardState,
GuesserMoveEvent,
NextStepRequestedEvent,
PlayerHelloEvent,
RevealSecretEvent,
SessionEvent,
SessionPhase,
SessionStats,
SetterMoveEvent,
StepData,
StepState,
)
@@ -16,7 +23,14 @@ __all__ = [
"ArrowDirection",
"CardData",
"CardState",
"GuesserMoveEvent",
"NextStepRequestedEvent",
"PlayerHelloEvent",
"RevealSecretEvent",
"SessionEvent",
"SessionPhase",
"SessionStats",
"SetterMoveEvent",
"StepData",
"StepState",
]
src/mind_reader/domain/common.py
+1 -1
View File
@@ -7,7 +7,7 @@ class PlayerRole(StrEnum):
GUESSER = auto()
@dataclass
@dataclass(frozen=True)
class Player:
name: str
role: PlayerRole
src/mind_reader/domain/session.py
+49
View File
@@ -2,6 +2,8 @@ from dataclasses import dataclass
from enum import Enum, StrEnum, auto
from typing import Optional
from .common import Player
class ArrowDirection(StrEnum):
LEFT = auto()
@@ -16,8 +18,18 @@ class StepState(Enum):
SESSION_FINISHED = auto()
class SessionPhase(Enum):
INITIALIZING = auto()
WAITING_FOR_SETTER = auto()
WAITING_FOR_GUESSER = auto()
WAITING_FOR_REVEAL = auto()
CHECKOUT = auto()
FINISHED = auto()
class CardState(Enum):
SETTER_NOT_SET = auto()
SETTER_WAITING = auto()
SETTER_HIDDEN = auto()
SETTER_REVEALED = auto()
GUESSER_DISABLED = auto()
@@ -54,3 +66,40 @@ class SessionStats:
accuracy = 0.0
accuracy = (self.correct / self.total) * 100
return round(accuracy, 1)
@dataclass(frozen=True)
class PlayerHelloEvent:
player: Player
@dataclass(frozen=True)
class SetterMoveEvent:
player: Player
direction: ArrowDirection
@dataclass(frozen=True)
class GuesserMoveEvent:
player: Player
direction: ArrowDirection
@dataclass(frozen=True)
class RevealSecretEvent:
player: Player
direction: ArrowDirection
@dataclass(frozen=True)
class NextStepRequestedEvent:
player: Player
SessionEvent = (
PlayerHelloEvent
| SetterMoveEvent
| GuesserMoveEvent
| RevealSecretEvent
| NextStepRequestedEvent
)
src/mind_reader/network/__init__.py
+10 -10
View File
@@ -1,11 +1,11 @@
from .controller import NetworkController
from .enums import NetworkMessageKind
from .handlers import SessionNetworkMessageHandler
from .message import NetworkMessage
# from .controller import NetworkController
# from .enums import NetworkMessageKind
# from .handlers import SessionNetworkMessageHandler
# from .message import NetworkMessage
__all__ = [
"NetworkController",
"NetworkMessageKind",
"SessionNetworkMessageHandler",
"NetworkMessage",
]
# __all__ = [
# "NetworkController",
# "NetworkMessageKind",
# "SessionNetworkMessageHandler",
# "NetworkMessage",
# ]
src/mind_reader/network/controller.py
+59 -41
View File
@@ -1,25 +1,42 @@
from typing import Callable
import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion
from .enums import NetworkMessageKind
from .message import NetworkMessage
MessageHandler = Callable[[NetworkMessage], None]
from ..state_machines.session.events import SessionEvent
from ..state_machines.session.machine import SessionStateMachine
from .message import EventMapper, NetworkMessage, SessionMessageKind
class NetworkController:
def __init__(self, host: str, port: int, username: str, password: str):
self._handlers: set[MessageHandler] = set()
def __init__(
self,
host: str,
port: int,
username: str,
password: str,
lobby,
session: SessionStateMachine,
):
self._lobby_topic = "mind_reader/lobby"
self._session_topic = "mind_reader/session"
self._lobby = lobby
self._session = session
self._pending_loopbacks: dict[str, NetworkMessage] = {}
self._topic = "mind_reader"
self._client = mqtt.Client(CallbackAPIVersion.VERSION2)
self._client.username_pw_set(username, password)
self._client.on_message = self._on_message
self._client.connect(host, port)
self._client.subscribe("mind_reader/#")
self._client.subscribe(self._lobby_topic)
self._client.subscribe(self._session_topic)
self._client.message_callback_add(
self._lobby_topic, self._on_lobby_message
)
self._client.message_callback_add(
self._session_topic, self._on_session_message
)
def start(self) -> None:
self._client.loop_start()
@@ -28,26 +45,12 @@ class NetworkController:
self._client.loop_stop()
self._client.disconnect()
def add_handler(self, handler: MessageHandler) -> None:
self._handlers.add(handler)
def remove_handler(self, handler: MessageHandler) -> None:
self._handlers.remove(handler)
def send(self, network_message: NetworkMessage) -> None:
kind = network_message.kind
payload = network_message.to_json()
match kind:
case (
NetworkMessageKind.SESSION_PLAYER_MOVED
| NetworkMessageKind.SESSION_SECRET_REVEAL_REQUESTED
| NetworkMessageKind.SESSION_SECRET_REVEALED
| NetworkMessageKind.SESSION_SESSION_FINISHED
| NetworkMessageKind.SESSION_SESSION_STARTED
):
topic = f"{self._topic}/session"
self._client.publish(topic, payload)
match network_message.kind:
case SessionMessageKind:
self._client.publish(
self._session_topic, network_message.to_json()
)
def send_with_looback(
self,
@@ -63,18 +66,33 @@ class NetworkController:
self._pending_loopbacks[local_message.id] = local_message
self.send(network_message)
def _handle(self, network_message: NetworkMessage) -> None:
for handler in self._handlers:
handler(network_message)
def _check_loopbacks(self, id: str) -> NetworkMessage | None:
if id in self._pending_loopbacks:
return self._pending_loopbacks.pop(id)
return None
def _on_message(
def _on_lobby_message(
self, client: mqtt.Client, userdata: None, message: mqtt.MQTTMessage
) -> None:
json_str = message.payload.decode("utf-8")
incoming_message = NetworkMessage.from_json(json_str)
json_str = message.payload.decode()
network_message = NetworkMessage.from_json(json_str)
if incoming_message.id in self._pending_loopbacks:
local_message = self._pending_loopbacks.pop(incoming_message.id)
self._handle(local_message)
else:
self._handle(incoming_message)
loopback_message = self._check_loopbacks(network_message.id)
if loopback_message:
network_message = loopback_message
event = EventMapper.from_message(network_message)
def _on_session_message(
self, client: mqtt.Client, userdata: None, message: mqtt.MQTTMessage
) -> None:
json_str = message.payload.decode()
network_message = NetworkMessage.from_json(json_str)
loopback_message = self._check_loopbacks(network_message.id)
if loopback_message:
network_message = loopback_message
event = EventMapper.from_message(network_message)
if isinstance(event, SessionEvent):
self._session.process(event)
src/mind_reader/network/enums.py
-15
View File
@@ -1,15 +0,0 @@
from enum import StrEnum
class NetworkMessageKind(StrEnum):
# ----- Lobby -----
# ---- Session ----
SESSION_SESSION_STARTED = "session_started"
SESSION_PLAYER_MOVED = "player_moved"
SESSION_SECRET_REVEAL_REQUESTED = "secret_reveal_requested"
SESSION_SECRET_REVEALED = "secret_revealed"
SESSION_SESSION_FINISHED = "session_finished"
# ---- System -----
SYSTEM_INC_MSG_PA_ERR = "incoming_message_parsing_error"
src/mind_reader/network/handlers.py
-47
View File
@@ -1,47 +0,0 @@
from ..domain import ArrowDirection, Player, PlayerRole
from ..state import SessionState
from .controller import NetworkController
from .enums import NetworkMessageKind
from .message import NetworkMessage
class SessionNetworkMessageHandler:
def __init__(self, network: NetworkController, state: SessionState):
self._network = network
self._state = state
def __call__(self, message: NetworkMessage):
match message.kind:
case NetworkMessageKind.SESSION_SESSION_STARTED:
self._state.start_session()
case NetworkMessageKind.SESSION_PLAYER_MOVED:
player_role = PlayerRole(message.payload["player"]["role"])
direction = ArrowDirection(message.payload["arrow_direction"])
match player_role:
case PlayerRole.SETTER:
self._state.setter_move(direction)
case PlayerRole.GUESSER:
self._state.guesser_move(direction)
case NetworkMessageKind.SESSION_SECRET_REVEAL_REQUESTED:
self._network.send(
NetworkMessage(
NetworkMessageKind.SESSION_SECRET_REVEALED,
{
"player": {
"name": self._state.local_player.name,
"role": self._state.local_player.role,
},
"arrow_direction": self._state.reveal_setter_move(),
},
)
)
case NetworkMessageKind.SESSION_SECRET_REVEALED:
direction = ArrowDirection(message.payload["arrow_direction"])
self._state.step_close(direction)
case NetworkMessageKind.SESSION_SESSION_FINISHED:
pass
src/mind_reader/network/message.py
+96 -16
View File
@@ -1,34 +1,114 @@
import json
from dataclasses import asdict, dataclass, field
from typing import Any, Optional
from enum import StrEnum, auto
from typing import Any
from uuid_extensions import uuid7str
from .enums import NetworkMessageKind
from ..state_machines.lobby.domain import Player, PlayerRole
from ..state_machines.session.domain import ArrowDirection
from ..state_machines.session.events import (
GuesserMoveEvent,
NextStepRequestedEvent,
PlayerHelloEvent,
RevealSecretEvent,
SessionEvent,
SetterMoveEvent,
)
class SessionMessageKind(StrEnum):
PLAYER_HELLO = auto()
SETTER_MOVED = auto()
GUESSER_MOVED = auto()
SECRET_REVEALED = auto()
NEXT_STEP_REQUESTED = auto()
NetworkMessageKind = SessionMessageKind
Event = SessionEvent
EVENT_TO_KIND_MAP: dict[type[Event], NetworkMessageKind] = {
PlayerHelloEvent: SessionMessageKind.PLAYER_HELLO,
SetterMoveEvent: SessionMessageKind.SETTER_MOVED,
GuesserMoveEvent: SessionMessageKind.GUESSER_MOVED,
RevealSecretEvent: SessionMessageKind.SECRET_REVEALED,
NextStepRequestedEvent: SessionMessageKind.NEXT_STEP_REQUESTED,
}
KIND_TO_EVENT_MAP = {
kind: event_cls for event_cls, kind in EVENT_TO_KIND_MAP.items()
}
@dataclass(frozen=True)
class NetworkMessage:
kind: NetworkMessageKind
payload: dict[str, Any]
payload: dict[str, Any] = field(default_factory=dict)
id: str = field(default_factory=uuid7str)
@classmethod
def from_json(cls, json_str: str) -> "NetworkMessage":
data = json.loads(json_str)
return cls(
kind=NetworkMessageKind(data["kind"]),
payload=data.get("payload", {}),
id=data["id"],
)
def to_json(self) -> str:
data = asdict(self)
data["kind"] = self.kind.value
return json.dumps(data)
return json.dumps(data, ensure_ascii=False)
class EventMapper:
@staticmethod
def to_message(event: Event) -> NetworkMessage:
event_type = type(event)
kind = EVENT_TO_KIND_MAP.get(event_type)
if not kind:
raise ValueError(f"Event {event_type} not registered")
return NetworkMessage(kind=kind, payload=asdict(event))
@staticmethod
def from_message(message: NetworkMessage) -> Event | None:
event_cls = KIND_TO_EVENT_MAP.get(message.kind)
if not event_cls:
return None
@classmethod
def from_json(cls, json_str: str) -> "NetworkMessage":
try:
data = json.loads(json_str)
return cls(
id=data["id"],
kind=NetworkMessageKind(data["kind"]),
payload=data.get("payload", {}),
)
except (json.JSONDecodeError, KeyError, ValueError) as e:
return cls(
kind=NetworkMessageKind.SYSTEM_INC_MSG_PA_ERR,
payload={"exception": e},
payload = message.payload
player = Player(
name=payload["player"]["name"],
role=PlayerRole(payload["player"]["role"]),
)
match message.kind:
case NetworkMessageKind.PLAYER_HELLO:
return PlayerHelloEvent(player=player)
case NetworkMessageKind.SETTER_MOVED:
return SetterMoveEvent(
player=player,
direction=ArrowDirection(payload["direction"]),
)
case NetworkMessageKind.GUESSER_MOVED:
return GuesserMoveEvent(
player=player,
direction=ArrowDirection(payload["direction"]),
)
case NetworkMessageKind.SECRET_REVEALED:
return RevealSecretEvent(
player=player,
direction=ArrowDirection(payload["direction"]),
)
case NetworkMessageKind.NEXT_STEP_REQUESTED:
return NextStepRequestedEvent(player=player)
except (KeyError, ValueError, TypeError):
return None
src/mind_reader/state_machines/__init__.py
View File
src/mind_reader/state_machines/lobby/__init__.py
View File
src/mind_reader/state_machines/lobby/domain.py
+13
View File
@@ -0,0 +1,13 @@
from dataclasses import dataclass
from enum import StrEnum, auto
class PlayerRole(StrEnum):
SETTER = auto()
GUESSER = auto()
@dataclass(frozen=True)
class Player:
name: str
role: PlayerRole
src/mind_reader/state_machines/session/__init__.py
View File
src/mind_reader/state_machines/session/domain.py
+27
View File
@@ -0,0 +1,27 @@
from enum import Enum, StrEnum, auto
class ArrowDirection(StrEnum):
LEFT = auto()
RIGHT = auto()
HIDDEN = auto()
class SessionPhase(Enum):
INITIALIZING = auto()
WAITING_FOR_SETTER = auto()
WAITING_FOR_GUESSER = auto()
WAITING_FOR_REVEAL = auto()
CHECKOUT = auto()
FINISHED = auto()
class CardState(Enum):
SETTER_NOT_SET = auto()
SETTER_WAITING = auto()
SETTER_HIDDEN = auto()
SETTER_REVEALED = auto()
GUESSER_DISABLED = auto()
GUESSER_WAITING = auto()
GUESSER_CORRECT = auto()
GUESSER_WRONG = auto()
src/mind_reader/state_machines/session/events.py
+41
View File
@@ -0,0 +1,41 @@
from dataclasses import dataclass
from ..lobby.domain import Player
from ..session.domain import ArrowDirection
@dataclass(frozen=True)
class PlayerHelloEvent:
player: Player
@dataclass(frozen=True)
class SetterMoveEvent:
player: Player
direction: ArrowDirection
@dataclass(frozen=True)
class GuesserMoveEvent:
player: Player
direction: ArrowDirection
@dataclass(frozen=True)
class RevealSecretEvent:
player: Player
direction: ArrowDirection
@dataclass(frozen=True)
class NextStepRequestedEvent:
player: Player
SessionEvent = (
PlayerHelloEvent
| SetterMoveEvent
| GuesserMoveEvent
| RevealSecretEvent
| NextStepRequestedEvent
)
src/mind_reader/state_machines/session/machine.py
+181
View File
@@ -0,0 +1,181 @@
from ..lobby.domain import Player, PlayerRole
from ..session.domain import ArrowDirection, CardState, SessionPhase
from ..session.events import (
GuesserMoveEvent,
NextStepRequestedEvent,
PlayerHelloEvent,
RevealSecretEvent,
SessionEvent,
SetterMoveEvent,
)
class SessionStateMachine:
def __init__(
self, setter: Player, guessers: set[Player], sequence_length: int = 20
):
self._setter = setter
self._guessers = guessers
self._pending_players: set[Player] = set(guessers)
self._pending_players.add(setter)
self._sequence_length = sequence_length
self._setter_moves: list[ArrowDirection] = []
self._guessers_moves: dict[Player, list[ArrowDirection]] = {
player: [] for player in guessers
}
self._phase: SessionPhase = SessionPhase.INITIALIZING
def process(self, event: SessionEvent) -> bool:
match (self._phase, event):
case (SessionPhase.INITIALIZING, PlayerHelloEvent(player)):
self._pending_players.remove(player)
if not self._pending_players:
self._phase = SessionPhase.WAITING_FOR_SETTER
self._pending_players = {self._setter}
return True
case (
SessionPhase.WAITING_FOR_SETTER,
SetterMoveEvent(player, direction),
):
self._pending_players.remove(player)
if not self._pending_players:
self._setter_moves.append(direction)
self._phase = SessionPhase.WAITING_FOR_GUESSER
self._pending_players = set(self._guessers)
return True
case (
SessionPhase.WAITING_FOR_GUESSER,
GuesserMoveEvent(player, direction),
):
self._guessers_moves[player].append(direction)
self._pending_players.remove(player)
if not self._pending_players:
self._phase = SessionPhase.WAITING_FOR_REVEAL
self._pending_players = {self._setter}
return True
case (
SessionPhase.WAITING_FOR_REVEAL,
RevealSecretEvent(player, direction),
):
self._pending_players.remove(player)
if not self._pending_players:
self._phase = SessionPhase.CHECKOUT
setter_move_index = len(self._setter_moves) - 1
self._setter_moves[setter_move_index] = direction
self._pending_players = set(self._guessers)
return True
case (SessionPhase.CHECKOUT, NextStepRequestedEvent(player)):
self._pending_players.remove(player)
if not self._pending_players:
if len(self._setter_moves) >= self._sequence_length:
self._phase = SessionPhase.FINISHED
self._pending_players.clear()
else:
self._phase = SessionPhase.WAITING_FOR_SETTER
self._pending_players = {self._setter}
return True
case _:
return False
def get_card_state(self, player: Player, index: int) -> CardState:
match player.role:
case PlayerRole.SETTER:
return self._get_setter_card_state(index)
case PlayerRole.GUESSER:
return self._get_guesser_card_state(player, index)
def _get_setter_card_state(self, index: int) -> CardState:
current_index = self.get_setter_step_number() - 1
if index > current_index:
return CardState.SETTER_NOT_SET
elif index == current_index:
match self._phase:
case SessionPhase.INITIALIZING:
return CardState.SETTER_NOT_SET
case SessionPhase.WAITING_FOR_SETTER:
return CardState.SETTER_WAITING
case (
SessionPhase.WAITING_FOR_GUESSER
| SessionPhase.WAITING_FOR_REVEAL
):
if self._setter_moves[-1] == ArrowDirection.HIDDEN:
return CardState.SETTER_HIDDEN
else:
return CardState.SETTER_REVEALED
case SessionPhase.CHECKOUT | SessionPhase.FINISHED:
return CardState.SETTER_REVEALED
else:
return CardState.SETTER_REVEALED
def _get_guesser_card_state(self, player: Player, index: int) -> CardState:
def _check_direction(
self: SessionStateMachine, player: Player, index: int
) -> CardState:
if self._setter_moves[index] == self._guessers_moves[player][index]:
return CardState.GUESSER_CORRECT
else:
return CardState.GUESSER_WRONG
current_index = self.get_guesser_step_number(player) - 1
if index > current_index:
return CardState.GUESSER_DISABLED
elif index == current_index:
match self._phase:
case (
SessionPhase.INITIALIZING | SessionPhase.WAITING_FOR_SETTER
):
return CardState.GUESSER_DISABLED
case (
SessionPhase.WAITING_FOR_GUESSER
| SessionPhase.WAITING_FOR_REVEAL
):
return CardState.GUESSER_WAITING
case SessionPhase.CHECKOUT | SessionPhase.FINISHED:
return _check_direction(self, player, index)
else:
return _check_direction(self, player, index)
def _get_step_number(self, enum) -> int:
return len(enum)
def get_setter_step_number(self) -> int:
return self._get_step_number(self._setter_moves)
def get_guesser_step_number(self, player: Player) -> int:
return self._get_step_number(self._guessers_moves[player])
def get_setter_move(self, index: int) -> ArrowDirection | None:
try:
return self._setter_moves[index]
except IndexError:
return None
def get_guesser_move(
self, player: Player, index: int
) -> ArrowDirection | None:
try:
return self._guessers_moves[player][index]
except IndexError:
return None
def get_last_setter_move(self) -> ArrowDirection | None:
index = self.get_setter_step_number() - 1
return self.get_setter_move(index)
def get_last_guesser_move(self, player: Player) -> ArrowDirection | None:
index = self.get_guesser_step_number(player) - 1
return self.get_guesser_move(player, index)