Vis

.gitignore

@@ -1,4 +1,7 @@
 __pycache__/
+*.ipynb
+example.db
+db.sqlite
 *.py[cod]
 *$py.class
 *.so

pyproject.toml

@@ -13,7 +13,8 @@ python-telegram-bot = "^20.7"
 pydantic = "^2.5.3"
 cairosvg = "^2.7.1"
 pydantic-settings = "^2.2.1"
-
+pandas = "^1.5.3"
+matplotlib = "^3.7.4"
 
 [build-system]
 requires = ["poetry-core"]

src/get_handlers.py

@@ -11,6 +11,7 @@ def get_handlers() -> tuple:
         CommandHandler("help", handlers.help),
         CommandHandler("game", handlers.game),
         CommandHandler("save", handlers.save),
+        CommandHandler("mystats", handlers.mystats),
         # Poll answer handler
         PollAnswerHandler(poll_callback_receiver),
     )

src/handlers/__init__.py

@@ -2,12 +2,14 @@
 from .help import help
 from .game import game
 from .save import save
+from .mystats import mystats
 from .error_handler import error_handler
 
 __all__ = [
     "start",
     "help",
     "game",
     "save",
+    "mystats",
     "error_handler",
 ]

src/handlers/mystats.py

@@ -0,0 +1,19 @@
+import asyncio
+
+from telegram import Update
+from telegram.ext import ContextTypes
+
+from src.services.draw_graphs import draw_user_winrate
+from src.config import AppConfig
+
+
+async def mystats(
+    update: Update, context: ContextTypes.DEFAULT_TYPE, config: AppConfig = AppConfig()
+) -> None:
+    # Draws personal stats of asking user
+    user_id = update.effective_user.id
+    chat_id = update.effective_chat.id
+    await context.bot.send_photo(chat_id=chat_id,
+                                 photo=await draw_user_winrate(user_id), 
+                                 disable_notification=True)
+    #await update.effective_message.delete()

src/services/db_service.py

@@ -125,6 +125,25 @@ async def delete_poll_result(poll_id: int, user_id: int) -> None:
         )
 
 
+async def fetch_user(username=None, id=None, first_name=None, full_name=None, last_name=None):
+    # returns full player info by given info
+    conditions = []
+    if not id is None:
+        conditions.append(f'id = {id.__repr__()}')
+    if not username is None:
+        conditions.append(f'username = {username.__repr__()}')
+    if not first_name is None:
+        conditions.append(f'first_name = {first_name.__repr__()}')
+    if not last_name is None:
+        conditions.append(f'last_name = {last_name.__repr__()}')
+    if not full_name is None:
+        conditions.append(f'full_name = {full_name.__repr__()}')
+    query = f"""SELECT * FROM players
+                WHERE {' AND '.join(conditions)};"""
+    result = await fetch_one(query, [])
+    return result
+
+
 async def fetch_poll_data(poll_id: int) -> Poll | None:
     sql = """SELECT id, message_id, chat_id, chat_name, creator_id, creator_username
              FROM polls
@@ -141,3 +160,140 @@ async def fetch_poll_results(poll_id: int) -> tuple[PollResult]:
              WHERE poll_id = ?"""
     results = await fetch_all(sql, [poll_id])
     return tuple(PollResult(**result) for result in results)
+
+
+async def fetch_player_answers(user_id):
+    """
+    Returns table of player with given username results grouped by his answers
+
+    Parameters:
+    -----------
+    cur : sqlite3 cursor
+        Cursor to the given database.
+
+    user_id : int
+        Id of a given player.
+
+    Returns:
+    --------
+    res : dict
+        Keys are answers, values are count numbers
+    """
+    query = f"""SELECT SUM(CASE WHEN records.role = 'HC' THEN 1 ELSE 0 END) AS HC, 
+                       SUM(CASE WHEN records.role = 'HD' THEN 1 ELSE 0 END) AS HD, 
+                       SUM(CASE WHEN records.role = 'HL' THEN 1 ELSE 0 END) AS HL, 
+                       SUM(CASE WHEN records.role = 'HW' THEN 1 ELSE 0 END) AS HW, 
+                       SUM(CASE WHEN records.role = 'FL' THEN 1 ELSE 0 END) AS FL, 
+                       SUM(CASE WHEN records.role = 'LL' THEN 1 ELSE 0 END) AS LL, 
+                       SUM(CASE WHEN records.role = 'LW' THEN 1 ELSE 0 END) AS LW, 
+                       SUM(CASE WHEN records.role = 'FW' THEN 1 ELSE 0 END) AS FW
+                FROM records
+                WHERE records.player_id = ?;"""
+    res = await fetch_one(query, [user_id])
+    return res
+
+
+async def fetch_players_stats(order='DESC', mingames=None, top=None):
+    """
+    Returns table containing number of wins and loses for each role and winrate
+
+    Parameters:
+    -----------
+    order : str values 'DESC' or 'ASC'
+
+    top : uint or None
+        function returns top number of results, if that's not None  
+
+    Retuens:
+    --------
+    res : DataFrame
+        Columns: username, 
+                 LW (liberal wins), 
+                 FW (fascist wins), 
+                 HW (Hitler wins), 
+                 LL (liberal loses), 
+                 FL (fascist loses), 
+                 HL (Hitler loses), 
+                 winrate
+    """
+    if mingames is None:
+        condition = ""
+    else:
+        condition = f"HAVING games >= {mingames}"
+    query = f"""SELECT players.id, players.username, players.full_name, 
+            SUM(CASE WHEN records.role = 'LW' THEN 1 ELSE 0 END) AS LW,
+            SUM(CASE WHEN records.role = 'FW' THEN 1 ELSE 0 END) AS FW,
+            SUM(CASE WHEN records.role IN ('HW', 'HC') THEN 1 ELSE 0 END) AS HW,
+            SUM(CASE WHEN records.role = 'LL' THEN 1 ELSE 0 END) AS LL,
+            SUM(CASE WHEN records.role = 'FL' THEN 1 ELSE 0 END) AS FL, 
+            SUM(CASE WHEN records.role IN ('HL', 'HD') THEN 1 ELSE 0 END) AS HL,
+            COUNT(records.role) AS games,
+            AVG(CASE WHEN records.role IN ('LW', 'FW', 'HW', 'HC') THEN 1 ELSE 0 END) AS winrate
+            FROM records
+            INNER JOIN players ON players.id = records.player_id
+            GROUP BY player_id {condition}
+            ORDER BY winrate {order};"""
+    if top is not None:
+        query = query[:-1] + f'\nLIMIT {top};'
+    res = await fetch_all(query, [])
+    return res
+
+
+async def fetch_connection_stats(username, order='DESC', top=None, which='teammate'):
+    """
+    Returns table containing number of wins and loses for each role and winrate
+
+    Parameters:
+    -----------
+    username : string from table Players.username
+        Username of given player.
+
+    order : str values 'DESC' or 'ASC'
+
+    top : uint or None
+        Function will return top n number of results, if that's not None  
+
+    which: str or None
+        Define which stats this function will return:
+        'teammate' : teammates stats, 
+        'opponent' : opponents stats,
+        None : full stats
+
+    Returns:
+    --------
+    res : DataFrame
+        Columns: username, 
+                 LW - Wins playing in liberal team
+                 LL - Loses playing in liberal team
+                 FW - Wins playing in fascist team
+                 FL - Loses playing in fascist team
+                 winrate
+    """
+    query_which = {None: '', 
+                   'teammate': "\nAND ((records.role in ('LW', 'LL') AND w.team = 'Liberal') OR (records.role IN ('FW', 'FL', 'HC', 'HL') AND w.team = 'Fascist'))", 
+                   'opponent': "\nAND ((records.role in ('LW', 'LL') AND w.team = 'Fascist') OR (records.role IN ('FW', 'FL', 'HC', 'HL') AND w.team = 'Liberal'))"
+                  }[which]
+    if top is None:
+        query_limit = ''
+    else:
+        query_limit = f'\nLIMIT {top}'
+    query = f"""WITH w(game_id, team, result) AS (SELECT records.game_id, 
+                CASE WHEN records.role IN ('LL', 'LW') THEN 'Liberal' ELSE 'Fascist' END AS team, 
+                CASE WHEN records.role IN ('FW', 'LW', 'HC', 'HW') THEN 'Win' ELSE 'Lose' END AS result
+                FROM records INNER JOIN players ON players.id = records.player_id 
+                WHERE players.username = ?)
+                SElECT players.username, players.full_name, 
+                SUM(CASE WHEN team = 'Liberal' AND result = 'Win' THEN 1 ELSE 0 END) AS LW, 
+                SUM(CASE WHEN team = 'Fascist' AND result = 'Win' THEN 1 ELSE 0 END) AS FW, 
+                SUM(CASE WHEN team = 'Liberal' AND result = 'Lose' THEN 1 ELSE 0 END) AS LL, 
+                SUM(CASE WHEN team = 'Fascist' AND result = 'Lose' THEN 1 ELSE 0 END) AS FL, 
+                AVG(CASE WHEN result = 'Win' THEN 1 ELSE 0 END) AS Winrate
+                FROM records 
+                INNER JOIN w ON w.game_id = records.game_id 
+                INNER JOIN players ON players.id = records.player_id
+                WHERE players.username != ? {query_which}
+                GROUP BY records.player_id
+                ORDER BY Winrate {order}{query_limit};"""
+
+    res = await fetch_all(query, [username, username])
+    return res

src/services/draw_graphs.py

@@ -0,0 +1,89 @@
+import io
+import asyncio
+import pandas as pd
+
+from matplotlib import pyplot as plt
+from matplotlib import image
+from matplotlib.offsetbox import (OffsetImage, AnnotationBbox)#The OffsetBox is a simple container artist.
+from telegram.ext import ContextTypes
+from src.services.db_service import fetch_user, fetch_player_answers, fetch_players_stats, fetch_connection_stats
+from src.services.draw_result_image import get_user_profile_photo, svg2png
+
+
+async def draw_user_winrate_bins(user_id, ax=None, return_bins=False):
+    """
+    Plots a winrate statistics for given player.
+
+    Parameters:
+    -----------
+    user_id : int from table Players.id
+        Id of a given player.
+
+    ax : matplotlib axes object, default None
+        An axes of the current figure
+
+    Returns:
+    --------
+        bins : pd.DataFrame
+    """
+    info = await fetch_player_answers(user_id)
+    bins = pd.DataFrame([[info['LW'], info['LL']], 
+                         [info['FW'], info['FL']], 
+                         [info['HC'] + info['HW'], info['HD'] + info['HL']]], 
+                        columns=['Wins', 'Loses'], 
+                        index=['Liberal', 'Fascist', 'Hitler']).transpose()
+
+    bins.plot(kind='bar', stacked=True, color=['deepskyblue', 'orangered', 'darkred'], rot='horizontal', ax=ax)
+    if return_bins:
+        return bins
+
+
+
+async def draw_user_winrate(user_id, outcome=None, context=ContextTypes.DEFAULT_TYPE):
+    """
+    Draw a winrate statistics for given player and saves this as svg
+
+    Parameters:
+    -----------
+    user_id : int from table Players.id
+        Id of a given player.
+
+    outcome: str
+        Outcome file name
+        Returns svg-string if this is None
+
+    context : telegram.ext.CallbackContext
+    """
+    user = await fetch_user(id=user_id)
+    print(user)
+
+    fig, ax = plt.subplots(1, 1)
+    fig.set_figwidth(8)
+    fig.set_figheight(6)
+    fig.suptitle(user['full_name'])
+
+    # Define a logo (an avatar) position
+    bins = await draw_user_winrate_bins(user_id, ax=ax, return_bins=True)
+    heights = bins.sum(axis=1)
+    if heights['Wins'] < 0.75*heights['Loses']:
+        ab_posx = 0
+    elif heights['Loses'] < 0.75*heights['Wins']:
+        ab_posx = 1
+    else:
+        ab_posx = 0.5
+    ab_posy = 0.85*heights.max()
+
+    # Set a logo (an avatar)
+    file = await get_user_profile_photo(context=context, player_id=user['id'])
+    logo = image.imread(file)
+    imagebox = OffsetImage(logo, zoom = 0.25)
+    ab = AnnotationBbox(imagebox, (ab_posx, ab_posy), frameon = True)
+    ax.add_artist(ab)
+
+    if not (outcome is None):
+        fig.savefig(outcome)
+    else:
+        svg = io.StringIO()
+        fig.savefig(svg, format='svg')
+        svg = svg.getvalue()
+        return svg2png(svg)