diff --git a/src/inventory_wars/game.py b/src/inventory_wars/game.py index bdc53fa..a844001 100644 --- a/src/inventory_wars/game.py +++ b/src/inventory_wars/game.py @@ -1,37 +1,24 @@ -from datetime import datetime - -from scapy.sendrecv import AsyncSniffer -from sqlalchemy.orm import Session - import logging -from star_resonance_tracer.proto.stru_place_holder_item_pb2 import PlaceHolderItem -from star_resonance_tracer.sniffer import Sniffer -from inventory_wars.models import User, ItemShare, Event -from inventory_wars.const import ChitChatNtf, HypertextVariant, decode_placeholder -from star_resonance_tracer.proto.serv_chit_chat_ntf_pb2 import ChitChatNtf as ChitChatNtfPb +from sqlalchemy.orm import Session from star_resonance_tracer.proto.enum_chit_chat_channel_type_pb2 import ChitChatChannelType from star_resonance_tracer.proto.enum_chit_chat_msg_type_pb2 import ChitChatMsgType +from star_resonance_tracer.proto.serv_chit_chat_ntf_pb2 import ChitChatNtf as ChitChatNtfPb +from star_resonance_tracer.sniffer import Sniffer +from inventory_wars.const import ChitChatNtf, HypertextVariant +from inventory_wars.scoring import Scoring from inventory_wars.sniffer import get_sniffer +from inventory_wars.state import State, GameIdle, GameOngoing logger = logging.getLogger(__name__) class Game: - def __init__(self, session: Session, *, item_id: int, listening_channels: list[ChitChatChannelType] | None = None): + def __init__(self, session: Session, *, listening_channels: list[ChitChatChannelType] | None = None): self.session = session self.listening_channels = listening_channels or [] - self.event = Event(item_id=item_id) - self.scapy: AsyncSniffer | None = None - - def start(self) -> None: - self.event.timestamp = datetime.now() - - self.session.add(self.event) - self.session.commit() - - logger.info(f"Started {self.event.id} at {self.event.timestamp} for item {self.event.item_id}") + self._state = GameIdle(self) sniffer = Sniffer() sniffer.set_service_type( @@ -44,8 +31,16 @@ class Game: self.scapy = get_sniffer(sniffer) self.scapy.start() - def stop(self): - self.scapy.stop() + def set_state(self, state: State) -> None: + self._state = state + + def start(self, item_id: int, scoring: Scoring) -> None: + if isinstance(self._state, GameIdle): + self._state.start(item_id, scoring) + + def end(self) -> None: + if isinstance(self._state, GameOngoing): + self._state.end() def on_chit_chat_msg(self, event: ChitChatNtfPb.NotifyNewestChitChatMsgs) -> None: req = event.vRequest @@ -59,25 +54,4 @@ class Game: if hypertext.configId != HypertextVariant.ITEM_SHARING.value: return - for placeholder in hypertext.hypertextContents: - placeholder_content = decode_placeholder(placeholder) - match placeholder_content: - case PlaceHolderItem() as item: - if item.configId != self.event.item_id: - continue - - user = User(id=req.chatMsg.sendCharInfo.charID, username=req.chatMsg.sendCharInfo.name) - item_share = ItemShare( - timestamp=datetime.fromtimestamp(req.chatMsg.timestamp), - user_id=user.id, - event_id=self.event.id, - count=item.ItemDetail.count, - raw=item.SerializeToString() - ) - - self.session.merge(user) - self.session.add(item_share) - self.session.commit() - - logger.info(f"{user.username} guessed {self.event.item_id} " - f"with {item_share.count} at {item_share.timestamp}") + self._state.handle_msg(req) diff --git a/src/inventory_wars/scoring.py b/src/inventory_wars/scoring.py new file mode 100644 index 0000000..832c436 --- /dev/null +++ b/src/inventory_wars/scoring.py @@ -0,0 +1,59 @@ +from typing import Protocol + +from inventory_wars.models import ItemShare + +__all__ = ( + "Scoring", + "FirstGuess", + "HighestAmount", + "FirstThenHighest" +) + + +class Scoring(Protocol): + def calculate_score(self, item: ItemShare) -> int: + raise NotImplementedError + + def calculate_end_score(self) -> ItemShare | None: + raise NotImplementedError + + +class FirstGuess(Scoring): + def __init__(self): + self.guessed = None + + def calculate_score(self, item: ItemShare) -> int: + if not self.guessed: + self.guessed = True + return 1 + return 0 + + def calculate_end_score(self) -> ItemShare | None: + return None + + +class HighestAmount(Scoring): + def __init__(self): + self.shares: set[ItemShare] = set() + + def calculate_score(self, item: ItemShare) -> int: + self.shares.add(item) + return 0 + + def calculate_end_score(self) -> ItemShare | None: + highest = max(self.shares, key=lambda s: s.count) + highest.score = 1 + return highest + + +class FirstThenHighest(Scoring): + def __init__(self): + self.first = FirstGuess() + self.highest = HighestAmount() + + def calculate_score(self, item: ItemShare) -> int: + self.highest.calculate_score(item) + return self.first.calculate_score(item) + + def calculate_end_score(self) -> ItemShare | None: + return self.highest.calculate_end_score() diff --git a/src/inventory_wars/state.py b/src/inventory_wars/state.py new file mode 100644 index 0000000..e9bc7cc --- /dev/null +++ b/src/inventory_wars/state.py @@ -0,0 +1,111 @@ +import logging +from datetime import datetime, UTC +from typing import TYPE_CHECKING, Protocol + +from star_resonance_tracer.proto.stru_notify_newest_chit_chat_msgs_request_pb2 import NotifyNewestChitChatMsgsRequest +from star_resonance_tracer.proto.stru_place_holder_item_pb2 import PlaceHolderItem + +from inventory_wars.const import decode_placeholder +from inventory_wars.models import Event, User, ItemShare +from inventory_wars.scoring import Scoring + +if TYPE_CHECKING: + from inventory_wars.game import Game + +__all__ = ( + "State", + "GameOngoing", + "GameIdle" +) + +logger = logging.getLogger(__name__) + + +class State(Protocol): + def start(self, item_id: int, scoring: type[Scoring]) -> None: + raise NotImplementedError + + def end(self) -> None: + raise NotImplementedError + + def handle_msg(self, msg: NotifyNewestChitChatMsgsRequest) -> None: + raise NotImplementedError + + +class GameIdle(State): + def __init__(self, game: Game): + self.game = game + + def start(self, item_id: int, scoring: Scoring) -> None: + event = Event(item_id=item_id) + + self.game.session.add(event) + self.game.session.commit() + + self.game.set_state(GameOngoing(self.game, scoring, event)) + + logger.info(f"Started {event.id} at {event.start_at} for item {event.item_id}.") + + def handle_msg(self, msg: NotifyNewestChitChatMsgsRequest) -> None: + hypertext = msg.chatMsg.msgInfo.chatHypertext + for placeholder in hypertext.hypertextContents: + placeholder_content = decode_placeholder(placeholder) + match placeholder_content: + case PlaceHolderItem() as item: + logger.info(item) + + +class GameOngoing(State): + def __init__(self, game: Game, scoring: Scoring, event: Event): + self.game = game + self.scoring = scoring + self.event = event + + def end(self): + self.event.end_at = datetime.now(UTC) + end = self.scoring.calculate_end_score() + if end: + end.user.score += end.score + self.game.session.commit() + + self.game.set_state(GameIdle(self.game)) + + logger.info(f"Ended {self.event.id} at {self.event.end_at}.") + if end: + logger.info(f"{end.user.username} with {end.item_id} at {end.count} earned {end.score} scores.") + + def handle_msg(self, msg: NotifyNewestChitChatMsgsRequest) -> None: + hypertext = msg.chatMsg.msgInfo.chatHypertext + for placeholder in hypertext.hypertextContents: + placeholder_content = decode_placeholder(placeholder) + match placeholder_content: + case PlaceHolderItem() as item: + user_id = msg.chatMsg.sendCharInfo.charID + user = self.game.session.get(User, user_id) + if user is None: + user = User(id=user_id, username=msg.chatMsg.sendCharInfo.name, score=0) + self.game.session.add(user) + + item_share = ItemShare( + timestamp=datetime.fromtimestamp(msg.chatMsg.timestamp, UTC), + user_id=user_id, + event_id=self.event.id if item.configId == self.event.item_id else None, + item_id=item.configId, + count=item.ItemDetail.count, + raw=item.SerializeToString() + ) + + if item_share.event_id: + item_share.score = self.scoring.calculate_score(item_share) + if item_share.score: + logger.info(user.score) + logger.info(item_share.score) + user.score += item_share.score + + self.game.session.add(item_share) + self.game.session.commit() + + logger.info(f"{user.username} guessed {item_share.item_id} " + f"with {item_share.count} at {item_share.timestamp}.") + if item_share.score: + logger.info(f"{user.username} earned {item_share.score} scores.")