Skip to content
9 changes: 8 additions & 1 deletion api.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@
}
STREAM_TIMEOUT = aiohttp.ClientTimeout(sock_connect=5.0, sock_read=9.0)


class API:
def __init__(self, config: Config) -> None:
self.lichess_session = aiohttp.ClientSession(
Expand Down Expand Up @@ -273,6 +272,7 @@ async def get_tournament_info(self, tournament_id: str) -> dict[str, Any]:
async with self.lichess_session.get(f"/api/tournament/{tournament_id}") as response:
return await response.json()


@retry(**JSON_RETRY_CONDITIONS)
async def get_user_status(self, username: str) -> dict[str, Any]:
async with self.lichess_session.get("/api/users/status", params={"ids": username}) as response:
Expand All @@ -289,6 +289,12 @@ async def handle_takeback(self, game_id: str, accept: bool) -> bool:
return False
return True

@retry(**JSON_RETRY_CONDITIONS)
async def get_user_activity(self, username: str) -> dict[str, Any]:
"""Get the user's recent activity"""
async with self.lichess_session.get(f"/api/user/{username}/activity") as response:
return await response.json()

@retry(**JSON_RETRY_CONDITIONS)
async def join_team(self, team: str, password: str | None) -> bool:
data = {"password": password} if password else None
Expand Down Expand Up @@ -382,3 +388,4 @@ async def withdraw_tournament(self, tournament_id: str) -> bool:
except aiohttp.ClientResponseError as e:
print(e)
return False

89 changes: 89 additions & 0 deletions user_interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
from enum import StrEnum
from typing import TypeVar

import psutil

from api import API
from botli_dataclasses import Challenge_Request
from config import Config
Expand All @@ -33,6 +35,7 @@
"quit": "Exits the bot.",
"rechallenge": "Challenges the opponent to the last received challenge.",
"reset": "Resets matchmaking. Usage: reset PERF_TYPE",
"stats": "Shows bot statistics and performance information.",
"stop": "Stops matchmaking mode.",
"tournament": "Joins tournament. Usage: tournament ID [TEAM_ID] [PASSWORD]",
"whitelist": "Temporarily whitelists a user. Use config for permanent whitelisting. Usage: whitelist USERNAME",
Expand Down Expand Up @@ -158,6 +161,8 @@ async def _handle_command(self, command: list[str]) -> None:
self._rechallenge()
case "reset":
self._reset(command)
case "stats":
await self._stats()
case "stop" | "s":
self._stop()
case "tournament" | "t":
Expand Down Expand Up @@ -286,6 +291,90 @@ def _reset(self, command: list[str]) -> None:
self.game_manager.matchmaking.opponents.reset_release_time(perf_type)
print("Matchmaking has been reset.")

async def _stats(self) -> None:
"""Display bot statistics and performance information."""
print("\n=== Bot Statistics ===")

try:
account = await self.api.get_account()
print(f"👤 Username: {account['username']}")

perfs = account.get('perfs', {})
if perfs:
print("\nCurrent Ratings:")
for perf_type, perf_data in perfs.items():
if 'rating' in perf_data:
rating = perf_data['rating']
provisional = "?" if perf_data.get('provisional', False) else ""
print(f" {perf_type.title():<15}: {rating}{provisional}")

activity = await self.api.get_user_activity(account['username'])
if activity and isinstance(activity, list) and len(activity) > 0:
today_activity = activity[0]

if 'games' in today_activity:
games = today_activity['games']
print("\nToday's Games:")

total_games = 0
try:
for variant_data in games.values():
if isinstance(variant_data, dict):
total_games += sum(count for count in variant_data.values() if isinstance(count, int))
except Exception:
total_games = 0

print(f"Total Games: {total_games}")

for variant, results in games.items():
if not isinstance(results, dict):
continue

wins = results.get('win', 0) if isinstance(results.get('win'), int) else 0
losses = results.get('loss', 0) if isinstance(results.get('loss'), int) else 0
draws = results.get('draw', 0) if isinstance(results.get('draw'), int) else 0
variant_total = wins + losses + draws

if variant_total > 0:
win_rate = (wins + 0.5 * draws) / variant_total * 100 if variant_total > 0 else 0

rating_diff = 0
if isinstance(results.get('rp'), dict):
rp_data = results['rp']
before = rp_data.get('before', 0)
after = rp_data.get('after', 0)
rating_diff = after - before

results_parts = []
if wins > 0:
results_parts.append(f"{wins} win{'s' if wins != 1 else ''}")
if draws > 0:
results_parts.append(f"{draws} draw{'s' if draws != 1 else ''}")
if losses > 0:
results_parts.append(f"{losses} loss{'es' if losses != 1 else ''}")

results_str = ", ".join(results_parts)
diff_str = f"+{rating_diff}" if rating_diff > 0 else str(rating_diff)
variant_name = variant.title()
print(f"{variant_name:<15}: ("
f"{variant_total} game{'s' if variant_total != 1 else ''}: "
f"{results_str} • {win_rate:.1f}% win rate • {diff_str})")

else:
print("\nNo games played today.")

except Exception as e:
print(f"Could not retrieve stats: {e}")

print("\nSystem Information:")
process = psutil.Process()
memory_mb = process.memory_info().rss / 1024 / 1024
print(f"Memory Usage: {memory_mb:.1f} MB")
cpu_percent = process.cpu_percent(interval=0.1)
print(f"CPU Usage: {cpu_percent:.1f}%")

print("=" * 25)

def _stop(self) -> None:
if self.game_manager.stop_matchmaking():
print("Stopping matchmaking ...")
Expand Down