Add chatterer

chatterer/__init__.py

@@ -0,0 +1 @@
+from .eventmanager import Eventmanager

chatterer/eventmanager.py

@@ -0,0 +1,21 @@
+from . import events
+import logging
+
+
+class Eventmanager:
+    def __init__(self, conversation):
+        self.conversation = conversation
+
+    def registered_events(self):
+        return events.BaseEvent.__subclasses__()
+
+    def handle_events(self, board, game):
+        logging.info(f"Checking events: {self.registered_events()}")
+        for event_class in self.registered_events():
+            event = event_class()
+            if event.is_triggering(board, game):
+                logging.info(f"triggering event {event}")
+                if event.is_sending_to_player():
+                    self.conversation.send_to_player(event.react_with(board, game))
+                else:
+                    self.conversation.send_to_spectator(event.react_with(board, game))

chatterer/events/__init__.py

@@ -0,0 +1,5 @@
+from .base_event import BaseEvent
+from .game_start import GameStartEvent
+from .game_end import GameEndEvent
+from .queen_won import QueenWonEvent
+from .queen_lost import QueenLostEvent

chatterer/events/base_event.py

@@ -0,0 +1,12 @@
+class BaseEvent:
+    def __init__(self):
+        pass
+
+    def is_triggering(self, board, game):
+        return False
+
+    def is_sending_to_player(self):
+        return True
+
+    def react_with(self, board, game):
+        return ""

chatterer/events/game_end.py

@@ -0,0 +1,13 @@
+from .base_event import BaseEvent
+import chess
+
+
+class GameEndEvent(BaseEvent):
+    def __init__(self):
+        pass
+
+    def is_triggering(self, board: chess.Board, game):
+        return board.is_game_over() or game.state["status"] != "started"
+
+    def react_with(self, board, game):
+        return f"This was a nice game. Feel free to challenge me anytime. Thanks for playing!"

chatterer/events/game_start.py

@@ -0,0 +1,13 @@
+from .base_event import BaseEvent
+import chess
+
+
+class GameStartEvent(BaseEvent):
+    def __init__(self):
+        pass
+
+    def is_triggering(self, board: chess.Board, game):
+        return len(board.move_stack) <= 1
+
+    def react_with(self, board, game):
+        return "Hello, nice to meet you! Have fun and good luck! :)"

chatterer/events/queen_lost.py

@@ -0,0 +1,20 @@
+from .base_event import BaseEvent
+import chess
+import logging
+
+
+class QueenLostEvent(BaseEvent):
+    def __init__(self):
+        pass
+
+    def is_triggering(self, board: chess.Board, game):
+        old_board = board.copy()
+        move = old_board.pop()
+        if len(old_board.pieces(chess.QUEEN, game.is_white)) > len(
+            board.pieces(chess.QUEEN, game.is_white)
+        ):
+            return True
+        return False
+
+    def react_with(self, board: chess.Board, game):
+        return f"(ノಠ益ಠ)ノ彡┻━┻ NO, MY QUEEN!"

chatterer/events/queen_won.py

@@ -0,0 +1,20 @@
+from .base_event import BaseEvent
+import chess
+import logging
+
+
+class QueenWonEvent(BaseEvent):
+    def __init__(self):
+        pass
+
+    def is_triggering(self, board: chess.Board, game):
+        old_board = board.copy()
+        move = old_board.pop()
+        if len(old_board.pieces(chess.QUEEN, not game.is_white)) > len(
+            board.pieces(chess.QUEEN, not game.is_white)
+        ):
+            return True
+        return False
+
+    def react_with(self, board: chess.Board, game):
+        return f"Ah, the Botez Gambit. Classic."

conversation.py

@@ -54,6 +54,12 @@ def command(self, line, game, cmd):
     def send_reply(self, line, reply):
         self.xhr.chat(self.game.id, line.room, reply)
 
+    def send_to_player(self, message):
+        self.xhr.chat(self.game.id, "player", message)
+
+    def send_to_spectator(self, message):
+        self.xhr.chat(self.game.id, "spectator", message)
+
 
 class ChatLine:
     def __init__(self, json):

lichess-bot.py

@@ -19,6 +19,7 @@
 from config import load_config
 from conversation import Conversation, ChatLine
 from functools import partial
+from chatterer.eventmanager import Eventmanager
 from requests.exceptions import (
     ChunkedEncodingError,
     ConnectionError,
@@ -239,6 +240,7 @@ def play_game(
     engine = engine_factory(board)
     engine.get_opponent_info(game)
     conversation = Conversation(game, engine, li, __version__, challenge_queue)
+    conversation_manager = Eventmanager(conversation)
 
     logger.info("+++ {}".format(game))
 
@@ -366,6 +368,7 @@ def ponder_thread_func(game, engine, board, wtime, btime, winc, binc):
                 moves = upd["moves"].split()
                 if len(moves) > 0 and len(moves) != len(board.move_stack):
                     board = update_board(board, moves[-1])
+                conversation_manager.handle_events(board, game)
                 if not is_game_over(game) and is_engine_move(game, moves):
                     if config.get("fake_think_time") and len(moves) > 9:
                         delay = (

model.py

@@ -11,35 +11,51 @@ def __init__(self, c_info):
         self.speed = c_info["speed"]
         self.increment = c_info.get("timeControl", {}).get("increment", -1)
         self.challenger = c_info.get("challenger")
-        self.challenger_title = self.challenger.get("title") if self.challenger else None
+        self.challenger_title = (
+            self.challenger.get("title") if self.challenger else None
+        )
         self.challenger_is_bot = self.challenger_title == "BOT"
-        self.challenger_master_title = self.challenger_title if not self.challenger_is_bot else None
-        self.challenger_name = self.challenger["name"] if self.challenger else "Anonymous"
+        self.challenger_master_title = (
+            self.challenger_title if not self.challenger_is_bot else None
+        )
+        self.challenger_name = (
+            self.challenger["name"] if self.challenger else "Anonymous"
+        )
         self.challenger_rating_int = self.challenger["rating"] if self.challenger else 0
         self.challenger_rating = self.challenger_rating_int or "?"
 
     def is_supported_variant(self, supported):
         return self.variant in supported
 
-    def is_supported_time_control(self, supported_speed, supported_increment_max, supported_increment_min):
+    def is_supported_time_control(
+        self, supported_speed, supported_increment_max, supported_increment_min
+    ):
         if self.increment < 0:
             return self.speed in supported_speed
-        return self.speed in supported_speed and self.increment <= supported_increment_max and self.increment >= supported_increment_min
+        return (
+            self.speed in supported_speed
+            and self.increment <= supported_increment_max
+            and self.increment >= supported_increment_min
+        )
 
     def is_supported_mode(self, supported):
         return "rated" in supported if self.rated else "casual" in supported
 
     def is_supported(self, config):
-        if ( not config.get("accept_bot", False) ) and self.challenger_is_bot:
+        if (not config.get("accept_bot", False)) and self.challenger_is_bot:
             return False
-        if config.get("only_bot", False) and ( not self.challenger_is_bot ):
+        if config.get("only_bot", False) and (not self.challenger_is_bot):
             return False
         variants = config["variants"]
         tc = config["time_controls"]
         inc_max = config.get("max_increment", 180)
         inc_min = config.get("min_increment", 0)
         modes = config["modes"]
-        return self.is_supported_time_control(tc, inc_max, inc_min) and self.is_supported_variant(variants) and self.is_supported_mode(modes)
+        return (
+            self.is_supported_time_control(tc, inc_max, inc_min)
+            and self.is_supported_variant(variants)
+            and self.is_supported_mode(modes)
+        )
 
     def score(self):
         rated_bonus = 200 if self.rated else 0
@@ -50,10 +66,18 @@ def mode(self):
         return "rated" if self.rated else "casual"
 
     def challenger_full_name(self):
-        return "{}{}".format(self.challenger_title + " " if self.challenger_title else "", self.challenger_name)
+        return "{}{}".format(
+            self.challenger_title + " " if self.challenger_title else "",
+            self.challenger_name,
+        )
 
     def __str__(self):
-        return "{} {} challenge from {}({})".format(self.perf_name, self.mode(), self.challenger_full_name(), self.challenger_rating)
+        return "{} {} challenge from {}({})".format(
+            self.perf_name,
+            self.mode(),
+            self.challenger_full_name(),
+            self.challenger_rating,
+        )
 
     def __repr__(self):
         return self.__str__()
@@ -65,23 +89,34 @@ def __init__(self, json, username, base_url, abort_time):
         self.id = json.get("id")
         self.speed = json.get("speed")
         clock = json.get("clock", {}) or {}
-        self.clock_initial = clock.get("initial", 1000 * 3600 * 24 * 365 * 10)  # unlimited = 10 years
+        self.clock_initial = clock.get(
+            "initial", 1000 * 3600 * 24 * 365 * 10
+        )  # unlimited = 10 years
         self.clock_increment = clock.get("increment", 0)
         self.perf_name = json.get("perf").get("name") if json.get("perf") else "{perf?}"
         self.variant_name = json.get("variant")["name"]
         self.white = Player(json.get("white"))
         self.black = Player(json.get("black"))
         self.initial_fen = json.get("initialFen")
         self.state = json.get("state")
-        self.is_white = bool(self.white.name and self.white.name.lower() == username.lower())
+        self.is_white = bool(
+            self.white.name and self.white.name.lower() == username.lower()
+        )
         self.my_color = "white" if self.is_white else "black"
         self.opponent_color = "black" if self.is_white else "white"
         self.me = self.white if self.is_white else self.black
         self.opponent = self.black if self.is_white else self.white
         self.base_url = base_url
-        self.white_starts = self.initial_fen == "startpos" or self.initial_fen.split()[1] == "w"
+        self.white_starts = (
+            self.initial_fen == "startpos" or self.initial_fen.split()[1] == "w"
+        )
         self.abort_at = time.time() + abort_time
-        self.terminate_at = time.time() + (self.clock_initial + self.clock_increment) / 1000 + abort_time + 60
+        self.terminate_at = (
+            time.time()
+            + (self.clock_initial + self.clock_increment) / 1000
+            + abort_time
+            + 60
+        )
 
     def url(self):
         return urljoin(self.base_url, "{}/{}".format(self.id, self.my_color))
@@ -124,7 +159,9 @@ def __str__(self):
             return "AI level {}".format(self.aiLevel)
         else:
             rating = "{}{}".format(self.rating, "?" if self.provisional else "")
-            return "{}{}({})".format(self.title + " " if self.title else "", self.name, rating)
+            return "{}{}({})".format(
+                self.title + " " if self.title else "", self.name, rating
+            )
 
     def __repr__(self):
         return self.__str__()