Replace ints and floats with datetime.timedeltas

conversation.py

@@ -5,6 +5,7 @@
 from engine_wrapper import EngineWrapper
 from lichess import Lichess
 from collections.abc import Sequence
+from timer import seconds
 MULTIPROCESSING_LIST_TYPE = Sequence[model.Challenge]
 
 logger = logging.getLogger(__name__)
@@ -53,7 +54,7 @@ def command(self, line: ChatLine, cmd: str) -> None:
         if cmd == "commands" or cmd == "help":
             self.send_reply(line, "Supported commands: !wait (wait a minute for my first move), !name, !howto, !eval, !queue")
         elif cmd == "wait" and self.game.is_abortable():
-            self.game.ping(60, 120, 120)
+            self.game.ping(seconds(60), seconds(120), seconds(120))
             self.send_reply(line, "Waiting 60 seconds...")
         elif cmd == "name":
             name = self.game.me.name

engine_wrapper.py

@@ -8,6 +8,7 @@
 import chess
 import subprocess
 import logging
+import datetime
 import time
 import random
 from collections import Counter
@@ -17,6 +18,7 @@
 import model
 import lichess
 from config import Configuration
+from timer import Timer, msec, seconds, msec_str, sec_str, to_seconds
 from typing import Any, Optional, Union
 OPTIONS_TYPE = dict[str, Any]
 MOVE_INFO_TYPE = dict[str, Any]
@@ -102,13 +104,13 @@ def play_move(self,
                   board: chess.Board,
                   game: model.Game,
                   li: lichess.Lichess,
-                  start_time: int,
-                  move_overhead: int,
+                  setup_timer: Timer,
+                  move_overhead: datetime.timedelta,
                   can_ponder: bool,
                   is_correspondence: bool,
-                  correspondence_move_time: int,
+                  correspondence_move_time: datetime.timedelta,
                   engine_cfg: config.Configuration,
-                  min_time: float) -> None:
+                  min_time: datetime.timedelta) -> None:
         """
         Play a move.
 
@@ -152,16 +154,16 @@ def play_move(self,
                 time_limit = first_move_time(game)
                 can_ponder = False  # No pondering after the first move since a new clock starts afterwards.
             elif is_correspondence:
-                time_limit = single_move_time(board, game, correspondence_move_time, start_time, move_overhead)
+                time_limit = single_move_time(board, game, correspondence_move_time, setup_timer, move_overhead)
             else:
-                time_limit = game_clock_time(board, game, start_time, move_overhead)
+                time_limit = game_clock_time(board, game, setup_timer, move_overhead)
 
             best_move = self.search(board, time_limit, can_ponder, draw_offered, best_move)
 
         # Heed min_time
-        elapsed = (time.perf_counter_ns() - start_time) / 1e9
+        elapsed = setup_timer.time_since_reset()
         if elapsed < min_time:
-            time.sleep(min_time - elapsed)
+            time.sleep(to_seconds(min_time - elapsed))
 
         self.add_comment(best_move, board)
         self.print_stats()
@@ -172,11 +174,11 @@ def play_move(self,
 
     def add_go_commands(self, time_limit: chess.engine.Limit) -> chess.engine.Limit:
         """Add extra commands to send to the engine. For example, to search for 1000 nodes or up to depth 10."""
-        movetime = self.go_commands.movetime
-        if movetime is not None:
-            movetime_sec = float(movetime) / 1000
-            if time_limit.time is None or time_limit.time > movetime_sec:
-                time_limit.time = movetime_sec
+        movetime_cfg = self.go_commands.movetime
+        if movetime_cfg is not None:
+            movetime = msec(movetime_cfg)
+            if time_limit.time is None or seconds(time_limit.time) > movetime:
+                time_limit.time = to_seconds(movetime)
         time_limit.depth = self.go_commands.depth
         time_limit.nodes = self.go_commands.nodes
         return time_limit
@@ -576,8 +578,8 @@ def getHomemadeEngine(name: str) -> type[MinimalEngine]:
     return engine
 
 
-def single_move_time(board: chess.Board, game: model.Game, search_time: int,
-                     start_time: int, move_overhead: int) -> chess.engine.Limit:
+def single_move_time(board: chess.Board, game: model.Game, search_time: datetime.timedelta,
+                     setup_timer: Timer, move_overhead: datetime.timedelta) -> chess.engine.Limit:
     """
     Calculate time to search in correspondence games.
 
@@ -588,13 +590,13 @@ def single_move_time(board: chess.Board, game: model.Game, search_time: int,
     :param move_overhead: The time it takes to communicate between the engine and lichess-bot.
     :return: The time to choose a move.
     """
-    pre_move_time = int((time.perf_counter_ns() - start_time) / 1e6)
+    pre_move_time = setup_timer.time_since_reset()
     overhead = pre_move_time + move_overhead
     wb = "w" if board.turn == chess.WHITE else "b"
-    clock_time = max(0, game.state[f"{wb}time"] - overhead)
+    clock_time = max(msec(0), msec(game.state[f"{wb}time"]) - overhead)
     search_time = min(search_time, clock_time)
-    logger.info(f"Searching for time {search_time} for game {game.id}")
-    return chess.engine.Limit(time=search_time / 1000, clock_id="correspondence")
+    logger.info(f"Searching for time {sec_str(search_time)} seconds for game {game.id}")
+    return chess.engine.Limit(time=to_seconds(search_time), clock_id="correspondence")
 
 
 def first_move_time(game: model.Game) -> chess.engine.Limit:
@@ -604,13 +606,16 @@ def first_move_time(game: model.Game) -> chess.engine.Limit:
     :param game: The game that the bot is playing.
     :return: The time to choose the first move.
     """
-    # Need to hardcode first movetime (10000 ms) since Lichess has 30 sec limit.
-    search_time = 10000
-    logger.info(f"Searching for time {search_time} for game {game.id}")
-    return chess.engine.Limit(time=search_time / 1000, clock_id="first move")
+    # Need to hardcode first movetime since Lichess has 30 sec limit.
+    search_time = seconds(10)
+    logger.info(f"Searching for time {sec_str(search_time)} seconds for game {game.id}")
+    return chess.engine.Limit(time=to_seconds(search_time), clock_id="first move")
 
 
-def game_clock_time(board: chess.Board, game: model.Game, start_time: int, move_overhead: int) -> chess.engine.Limit:
+def game_clock_time(board: chess.Board,
+                    game: model.Game,
+                    setup_timer: Timer,
+                    move_overhead: datetime.timedelta) -> chess.engine.Limit:
     """
     Get the time to play by the engine in realtime games.
 
@@ -620,15 +625,16 @@ def game_clock_time(board: chess.Board, game: model.Game, start_time: int, move_
     :param move_overhead: The time it takes to communicate between the engine and lichess-bot.
     :return: The time to play a move.
     """
-    pre_move_time = int((time.perf_counter_ns() - start_time) / 1e6)
+    pre_move_time = setup_timer.time_since_reset()
     overhead = pre_move_time + move_overhead
+    times = {side: msec(game.state[side]) for side in ["wtime", "btime"]}
     wb = "w" if board.turn == chess.WHITE else "b"
-    game.state[f"{wb}time"] = max(0, game.state[f"{wb}time"] - overhead)
-    logger.info("Searching for wtime {wtime} btime {btime}".format_map(game.state) + f" for game {game.id}")
-    return chess.engine.Limit(white_clock=game.state["wtime"] / 1000,
-                              black_clock=game.state["btime"] / 1000,
-                              white_inc=game.state["winc"] / 1000,
-                              black_inc=game.state["binc"] / 1000,
+    times[f"{wb}time"] = max(msec(0), times[f"{wb}time"] - overhead)
+    logger.info(f"Searching for wtime {msec_str(times['wtime'])} btime {msec_str(times['btime'])} for game {game.id}")
+    return chess.engine.Limit(white_clock=to_seconds(times["wtime"]),
+                              black_clock=to_seconds(times["btime"]),
+                              white_inc=to_seconds(msec(game.state["winc"])),
+                              black_inc=to_seconds(msec(game.state["binc"])),
                               clock_id="real time")
 
 
@@ -733,8 +739,8 @@ def get_chessdb_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
     """Get a move from chessdb.cn's opening book."""
     wb = "w" if board.turn == chess.WHITE else "b"
     use_chessdb = chessdb_cfg.enabled
-    time_left = game.state[f"{wb}time"]
-    min_time = chessdb_cfg.min_time * 1000
+    time_left = msec(game.state[f"{wb}time"])
+    min_time = seconds(chessdb_cfg.min_time)
     if not use_chessdb or time_left < min_time or board.uci_variant != "chess":
         return None, None
 
@@ -774,8 +780,8 @@ def get_lichess_cloud_move(li: lichess.Lichess, board: chess.Board, game: model.
                            lichess_cloud_cfg: config.Configuration) -> tuple[Optional[str], Optional[chess.engine.InfoDict]]:
     """Get a move from the lichess's cloud analysis."""
     wb = "w" if board.turn == chess.WHITE else "b"
-    time_left = game.state[f"{wb}time"]
-    min_time = lichess_cloud_cfg.min_time * 1000
+    time_left = msec(game.state[f"{wb}time"])
+    min_time = seconds(lichess_cloud_cfg.min_time)
     use_lichess_cloud = lichess_cloud_cfg.enabled
     if not use_lichess_cloud or time_left < min_time:
         return None, None
@@ -829,8 +835,8 @@ def get_opening_explorer_move(li: lichess.Lichess, board: chess.Board, game: mod
                               ) -> tuple[Optional[str], Optional[chess.engine.InfoDict]]:
     """Get a move from lichess's opening explorer."""
     wb = "w" if board.turn == chess.WHITE else "b"
-    time_left = game.state[f"{wb}time"]
-    min_time = opening_explorer_cfg.min_time * 1000
+    time_left = msec(game.state[f"{wb}time"])
+    min_time = seconds(opening_explorer_cfg.min_time)
     source = opening_explorer_cfg.source
     if not opening_explorer_cfg.enabled or time_left < min_time or source == "master" and board.uci_variant != "chess":
         return None, None
@@ -885,9 +891,9 @@ def get_online_egtb_move(li: lichess.Lichess, board: chess.Board, game: model.Ga
     wb = "w" if board.turn == chess.WHITE else "b"
     pieces = chess.popcount(board.occupied)
     source = online_egtb_cfg.source
-    minimum_time = online_egtb_cfg.min_time * 1000
+    minimum_time = seconds(online_egtb_cfg.min_time)
     if (not use_online_egtb
-            or game.state[f"{wb}time"] < minimum_time
+            or msec(game.state[f"{wb}time"]) < minimum_time
             or board.uci_variant not in ["chess", "antichess", "atomic"]
             and source == "lichess"
             or board.uci_variant != "chess"

lichess-bot.py

@@ -13,6 +13,7 @@
 import matchmaking
 import signal
 import time
+import datetime
 import backoff
 import os
 import io
@@ -23,7 +24,7 @@
 import traceback
 from config import load_config, Configuration
 from conversation import Conversation, ChatLine
-from timer import Timer
+from timer import Timer, seconds, msec, hours, to_seconds
 from requests.exceptions import ChunkedEncodingError, ConnectionError, HTTPError, ReadTimeout
 from asyncio.exceptions import TimeoutError as MoveTimeout
 from rich.logging import RichHandler
@@ -105,14 +106,14 @@ def watch_control_stream(control_queue: CONTROL_QUEUE_TYPE, li: lichess.Lichess)
     control_queue.put_nowait({"type": "terminated", "error": error})
 
 
-def do_correspondence_ping(control_queue: CONTROL_QUEUE_TYPE, period: int) -> None:
+def do_correspondence_ping(control_queue: CONTROL_QUEUE_TYPE, period: datetime.timedelta) -> None:
     """
     Tell the engine to check the correspondence games.
 
     :param period: How many seconds to wait before sending a correspondence ping.
     """
     while not terminated:
-        time.sleep(period)
+        time.sleep(to_seconds(period))
         control_queue.put_nowait({"type": "correspondence_ping"})
 
 
@@ -223,7 +224,7 @@ def start(li: lichess.Lichess, user_profile: USER_PROFILE_TYPE, config: Configur
     control_stream.start()
     correspondence_pinger = multiprocessing.Process(target=do_correspondence_ping,
                                                     args=(control_queue,
-                                                          config.correspondence.checkin_period))
+                                                          seconds(config.correspondence.checkin_period)))
     correspondence_pinger.start()
     correspondence_queue: CORRESPONDENCE_QUEUE_TYPE = manager.Queue()
 
@@ -301,7 +302,7 @@ def lichess_bot_main(li: lichess.Lichess,
                        if game["gameId"] not in startup_correspondence_games)
     low_time_games: list[EVENT_GETATTR_GAME_TYPE] = []
 
-    last_check_online_time = Timer(60 * 60)  # one hour interval
+    last_check_online_time = Timer(hours(1))
     matchmaker = matchmaking.Matchmaking(li, config, user_profile)
     matchmaker.show_earliest_challenge_time()
 
@@ -571,7 +572,7 @@ def play_game(li: lichess.Lichess,
     # Initial response of stream will be the full game info. Store it.
     initial_state = json.loads(next(lines).decode("utf-8"))
     logger.debug(f"Initial state: {initial_state}")
-    abort_time = config.abort_time
+    abort_time = seconds(config.abort_time)
     game = model.Game(initial_state, user_profile["username"], li.baseUrl, abort_time)
 
     with engine_wrapper.create_engine(config) as engine:
@@ -583,22 +584,22 @@ def play_game(li: lichess.Lichess,
 
         is_correspondence = game.speed == "correspondence"
         correspondence_cfg = config.correspondence
-        correspondence_move_time = correspondence_cfg.move_time * 1000
-        correspondence_disconnect_time = correspondence_cfg.disconnect_time
+        correspondence_move_time = seconds(correspondence_cfg.move_time)
+        correspondence_disconnect_time = seconds(correspondence_cfg.disconnect_time)
 
         engine_cfg = config.engine
         ponder_cfg = correspondence_cfg if is_correspondence else engine_cfg
         can_ponder = ponder_cfg.uci_ponder or ponder_cfg.ponder
-        move_overhead = config.move_overhead
-        delay_seconds = config.rate_limiting_delay / 1000
+        move_overhead = msec(config.move_overhead)
+        delay = msec(config.rate_limiting_delay)
 
         keyword_map: defaultdict[str, str] = defaultdict(str, me=game.me.name, opponent=game.opponent.name)
         hello = get_greeting("hello", config.greeting, keyword_map)
         goodbye = get_greeting("goodbye", config.greeting, keyword_map)
         hello_spectators = get_greeting("hello_spectators", config.greeting, keyword_map)
         goodbye_spectators = get_greeting("goodbye_spectators", config.greeting, keyword_map)
 
-        disconnect_time = correspondence_disconnect_time if not game.state.get("moves") else 0
+        disconnect_time = correspondence_disconnect_time if not game.state.get("moves") else seconds(0)
         prior_game = None
         board = chess.Board()
         upd: dict[str, Any] = game.state
@@ -615,28 +616,28 @@ def play_game(li: lichess.Lichess,
                     if not is_game_over(game) and is_engine_move(game, prior_game, board):
                         disconnect_time = correspondence_disconnect_time
                         say_hello(conversation, hello, hello_spectators, board)
-                        start_time = time.perf_counter_ns()
+                        setup_timer = Timer()
                         print_move_number(board)
                         move_attempted = True
                         engine.play_move(board,
                                          game,
                                          li,
-                                         start_time,
+                                         setup_timer,
                                          move_overhead,
                                          can_ponder,
                                          is_correspondence,
                                          correspondence_move_time,
                                          engine_cfg,
                                          fake_think_time(config, board, game))
-                        time.sleep(delay_seconds)
+                        time.sleep(to_seconds(delay))
                     elif is_game_over(game):
                         tell_user_game_result(game, board)
                         engine.send_game_result(game, board)
                         conversation.send_message("player", goodbye)
                         conversation.send_message("spectator", goodbye_spectators)
 
                     wb = "w" if board.turn == chess.WHITE else "b"
-                    terminate_time = (upd[f"{wb}time"] + upd[f"{wb}inc"]) / 1000 + 60
+                    terminate_time = msec(upd[f"{wb}time"]) + msec(upd[f"{wb}inc"]) + seconds(60)
                     game.ping(abort_time, terminate_time, disconnect_time)
                     prior_game = copy.deepcopy(game)
                 elif u_type == "ping" and should_exit_game(board, game, prior_game, li, is_correspondence):
@@ -672,12 +673,12 @@ def say_hello(conversation: Conversation, hello: str, hello_spectators: str, boa
         conversation.send_message("spectator", hello_spectators)
 
 
-def fake_think_time(config: Configuration, board: chess.Board, game: model.Game) -> float:
+def fake_think_time(config: Configuration, board: chess.Board, game: model.Game) -> datetime.timedelta:
     """Calculate how much time we should wait for fake_think_time."""
-    sleep = 0.0
+    sleep = seconds(0.0)
 
     if config.fake_think_time and len(board.move_stack) > 9:
-        remaining = max(0, game.my_remaining_seconds() - config.move_overhead / 1000)
+        remaining = max(seconds(0), game.my_remaining_time() - msec(config.move_overhead))
         delay = remaining * 0.025
         accel = 0.99 ** (len(board.move_stack) - 10)
         sleep = delay * accel

lichess.py

@@ -8,7 +8,8 @@
 import logging
 import traceback
 from collections import defaultdict
-from timer import Timer
+import datetime
+from timer import Timer, seconds, sec_str
 from typing import Optional, Union, Any
 import chess.engine
 JSON_REPLY_TYPE = dict[str, Any]
@@ -131,7 +132,7 @@ def api_get(self, endpoint_name: str, *template_args: str,
         response = self.session.get(url, params=params, timeout=timeout, stream=stream)
 
         if is_new_rate_limit(response):
-            delay = 1 if endpoint_name == "move" else 60
+            delay = seconds(1 if endpoint_name == "move" else 60)
             self.set_rate_limit_delay(path_template, delay)
 
         response.raise_for_status()
@@ -213,7 +214,7 @@ def api_post(self,
         response = self.session.post(url, data=data, headers=headers, params=params, json=payload, timeout=2)
 
         if is_new_rate_limit(response):
-            self.set_rate_limit_delay(path_template, 60)
+            self.set_rate_limit_delay(path_template, seconds(60))
 
         if raise_for_status:
             response.raise_for_status()
@@ -231,10 +232,10 @@ def get_path_template(self, endpoint_name: str) -> str:
         path_template = ENDPOINTS[endpoint_name]
         if self.is_rate_limited(path_template):
             raise RateLimited(f"{path_template} is rate-limited. "
-                              f"Will retry in {int(self.rate_limit_time_left(path_template))} seconds.")
+                              f"Will retry in {sec_str(self.rate_limit_time_left(path_template))} seconds.")
         return path_template
 
-    def set_rate_limit_delay(self, path_template: str, delay_time: int) -> None:
+    def set_rate_limit_delay(self, path_template: str, delay_time: datetime.timedelta) -> None:
         """
         Set a delay to a path template if it was rate limited.
 
@@ -248,7 +249,7 @@ def is_rate_limited(self, path_template: str) -> bool:
         """Check if a path template is rate limited."""
         return not self.rate_limit_timers[path_template].is_expired()
 
-    def rate_limit_time_left(self, path_template: str) -> float:
+    def rate_limit_time_left(self, path_template: str) -> datetime.timedelta:
         """How much time is left until we can use the path template normally."""
         return self.rate_limit_timers[path_template].time_until_expiration()