feat: improve game logic and add scoring

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 sqlalchemy.orm import Session
-from inventory_wars.const import ChitChatNtf, HypertextVariant, decode_placeholder
-from star_resonance_tracer.proto.serv_chit_chat_ntf_pb2 import ChitChatNtf as ChitChatNtfPb
 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._state = GameIdle(self)
-        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}")
 
         sniffer = Sniffer()
         sniffer.set_service_type(
@@ -44,8 +31,16 @@ class Game:
         self.scapy = get_sniffer(sniffer)
         self.scapy.start()
 
-    def stop(self):
+    def set_state(self, state: State) -> None:
-        self.scapy.stop()
+        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:
+        self._state.handle_msg(req)
-            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}")

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()

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.")