feat: Implemented League Of Legends

.gitignore

@@ -28,3 +28,4 @@ backup/
 .coverage
 
 dataset-values.json
+**/league-images/

.pre-commit-config.yaml

@@ -19,6 +19,7 @@ repos:
         exclude: ^(legacy-backend|tests)/.*$
         files: ^crispy-api/api/.*\.py$
         stages: [commit]
+        additional_dependencies: ["types-requests"]
   - repo: https://github.com/psf/black
     rev: 22.10.0
     hooks:

crispy-api/.gitignore

@@ -7,3 +7,5 @@ dataset/*
 .dataset/
 
 merged*
+
+.venv

crispy-api/api/__init__.py

@@ -23,8 +23,10 @@
     VIDEOS,
 )
 from api.tools.AI.network import NeuralNetwork
+from api.tools.enums import SupportedGames
 from api.tools.filters import apply_filters  # noqa
 from api.tools.setup import handle_highlights, handle_musics
+from api.tools.utils import download_champion_images
 
 ENCODERS_BY_TYPE[ObjectId] = str
 
@@ -40,7 +42,9 @@
 
 def init_app(debug: bool) -> FastAPI:
     if debug:
-        logging.getLogger("uvicorn").setLevel(logging.DEBUG)
+        logger = logging.getLogger("uvicorn")
+        logger.setLevel(logging.DEBUG)
+        logger.debug("Crispy started in debug mode")
         return FastAPI(debug=True)
     return FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
 
@@ -71,6 +75,8 @@ def is_tool_installed(ffmpeg_tool: str) -> None:
 async def setup_crispy() -> None:
     await handle_musics(MUSICS)
     await handle_highlights(VIDEOS, GAME, framerate=FRAMERATE)
+    if GAME == SupportedGames.LEAGUE_OF_LEGENDS:
+        await download_champion_images()
 
 
 @app.exception_handler(HTTPException)
@@ -87,4 +93,9 @@ def http_exception(request: Request, exc: HTTPException) -> JSONResponse:
 )
 
 
+@app.get("/health")
+def health() -> dict:
+    return {"status": "ok"}
+
+
 from api.routes import filters, highlight, music, result, segment  # noqa

crispy-api/api/config.py

@@ -1,11 +1,14 @@
 import json
 import os
+import warnings
 
 import easyocr
 from starlette.config import Config
 
 from api.tools.enums import SupportedGames
 
+warnings.filterwarnings("ignore", category=FutureWarning)
+
 config = Config(".env")
 
 DEBUG = config("DEBUG", cast=bool, default=False)
@@ -25,6 +28,8 @@
 VIDEOS = os.path.join(RESOURCES, "videos")
 MUSICS = os.path.join(RESOURCES, "musics")
 
+LEAGUE_IMAGES_PATH = os.path.join(os.getcwd(), "league-images")
+
 DATASET_PATH = "dataset"
 DATASET_VALUES_PATH = os.path.join(DATASET_PATH, "dataset-values.json")
 DATASET_CSV_PATH = os.path.join(DATASET_PATH, "result.csv")
@@ -43,7 +48,7 @@
 
     __clip = __settings.get("clip")
     if __clip is None:
-        raise KeyError(f"clips not found in {SETTINGS_JSON}")
+        raise KeyError(f"No clips in the {SETTINGS_JSON}")
 
     FRAMERATE = __clip.get("framerate", 8)
     OFFSET = __clip.get("second-between-kills", 0) * FRAMERATE
@@ -53,10 +58,13 @@
     GAME = __settings.get("game")
     if GAME is None:
         raise KeyError("game not found in settings.json")
-    if GAME.upper() not in [game.name for game in SupportedGames]:
+    if GAME.upper().replace("-", "_") not in [game.name for game in SupportedGames]:
         raise ValueError(f"game {GAME} not supported")
 
-    USE_NETWORK = GAME not in [SupportedGames.THEFINALS]
+    USE_NETWORK = GAME not in [
+        SupportedGames.THE_FINALS,
+        SupportedGames.LEAGUE_OF_LEGENDS,
+    ]
 
     __neural_network = __settings.get("neural-network")
     if __neural_network is None and USE_NETWORK:

crispy-api/api/models/highlight.py

@@ -298,6 +298,18 @@ def post_process(image: Image) -> Image:
             and killfeed_state
         )
 
+    async def extract_league_of_legends_images(
+        self, framerate: int = 4, stretch: bool = False
+    ) -> bool:
+        def post_process(image: Image) -> Image:
+            return image
+
+        return await self.extract_images(
+            post_process,
+            Box(1795, 240, 133, 245, 0, stretch, from_center=False),
+            framerate=framerate,
+        )
+
     async def extract_images_from_game(
         self, game: SupportedGames, framerate: int = 4, stretch: bool = False
     ) -> bool:
@@ -307,8 +319,10 @@ async def extract_images_from_game(
             return await self.extract_valorant_images(framerate, stretch)
         elif game == SupportedGames.CSGO2:
             return await self.extract_csgo2_images(framerate, stretch)
-        elif game == SupportedGames.THEFINALS:
+        elif game == SupportedGames.THE_FINALS:
             return await self.extract_the_finals_images(framerate, stretch)
+        elif game == SupportedGames.LEAGUE_OF_LEGENDS:
+            return await self.extract_league_of_legends_images(framerate, stretch)
         else:
             raise NotImplementedError(f"game {game} not supported")
 
@@ -458,9 +472,7 @@ async def scale_video(
         if not os.path.exists(self.path):
             raise FileNotFoundError(f"{self.path} not found")
 
-        logger.warning(
-            f"WARNING:Scaling video {self.path}, saving a backup in ./backup"
-        )
+        logger.warning(f"Scaling video {self.path}, saving a backup in ./backup")
 
         if not os.path.exists(backup):
             os.makedirs(backup)

crispy-api/api/tools/enums.py

@@ -5,4 +5,5 @@ class SupportedGames(str, Enum):
     VALORANT = "valorant"
     OVERWATCH = "overwatch"
     CSGO2 = "csgo2"
-    THEFINALS = "thefinals"
+    THE_FINALS = "the-finals"
+    LEAGUE_OF_LEGENDS = "league-of-legends"

crispy-api/api/tools/setup.py

@@ -78,7 +78,7 @@ def handle_specific_game(
     game: SupportedGames,
     framerate: int = 4,
 ) -> None:
-    if game == SupportedGames.THEFINALS:
+    if game == SupportedGames.THE_FINALS:
         handle_the_finals(new_highlights, framerate)
 
 

crispy-api/api/tools/utils.py

@@ -1,8 +1,81 @@
-from typing import Any, Dict, List
+import asyncio
+import logging
+import os
+from io import BytesIO
+from typing import Any, Dict, List, Union
 
+import aiohttp
+import requests
+from PIL import Image
+
+from api.config import LEAGUE_IMAGES_PATH
 from api.models.highlight import Highlight
 from api.tools.job_scheduler import JobScheduler
 
+logger = logging.getLogger("uvicorn")
+
+# URL for champion data and image base
+base_url = "https://ddragon.leagueoflegends.com"
+version_url = f"{base_url}/realms/na.json"
+json_url = f"{base_url}/cdn/VERSION/data/en_US/champion.json"
+image_base_url = f"{base_url}/cdn/VERSION/img/champion"
+
+
+def get_league_of_legends_version() -> Union[str, None]:
+    response = requests.get(version_url)
+    if response.status_code != 200:  # pragma: no cover
+        logger.error("Cannot download the league of legends patch version")
+        return None
+    data = response.json()
+    if "n" not in data or "champion" not in data["n"]:  # pragma: no cover
+        logger.error("Could not extract the league of legends patch version")
+    return str(data["n"]["champion"])
+
+
+async def fetch_image(session: aiohttp.ClientSession, url: str, path: str) -> None:
+    async with session.get(url) as response:
+        if response.status != 200:  # pragma: no cover
+            logger.error(f"Cannot download image from {url}")
+            return
+        img = Image.open(BytesIO(await response.read()))
+        img.save(path)
+
+
+async def download_champion_images(path: str = LEAGUE_IMAGES_PATH) -> None:
+    version = get_league_of_legends_version()
+    if not version:  # pragma: no cover
+        logger.error("Could not download the league of legends images")
+        return
+    async with aiohttp.ClientSession() as session:
+        champions = await session.get(json_url.replace("VERSION", version))
+        if champions.status != 200:  # pragma: no cover
+            logger.error("Cannot download the league of legends champion.json")
+            return
+        champion_names = (await champions.json())["data"].keys()
+
+        if not os.path.exists(path):  # pragma: no cover
+            os.makedirs(path)
+
+        logger.info("Downloading league of legends champion images")
+        tasks = []
+        for champion in champion_names:
+            image_path = os.path.join(path, f"{champion}.png")
+            if os.path.exists(image_path):
+                continue
+            image_url = f"{image_base_url.replace('VERSION', version)}/{champion}.png"
+            tasks.append(fetch_image(session, image_url, image_path))
+
+        await asyncio.gather(*tasks)
+
+    for champion in champion_names:
+        image_path = os.path.join(path, f"{champion}.png")
+        if os.path.exists(image_path):
+            img = Image.open(image_path)
+            img = img.resize((41, 41), Image.ANTIALIAS)
+            img.save(image_path)
+
+    logger.info("Done downloading and resizing league champions")
+
 
 def get_all_jobs_from_highlights(
     job_scheduler: JobScheduler, highlights: List[Highlight]

crispy-api/api/tools/video.py

@@ -2,12 +2,13 @@
 import logging
 import os
 from collections import Counter
-from typing import List, Tuple
+from typing import Any, Dict, List, Tuple, Union
 
+import cv2
 import numpy as np
 from PIL import Image
 
-from api.config import GAME, READER
+from api.config import GAME, LEAGUE_IMAGES_PATH, READER
 from api.models.highlight import Highlight
 from api.models.segment import Segment
 from api.tools.AI.network import NeuralNetwork
@@ -16,6 +17,8 @@
 
 logger = logging.getLogger("uvicorn")
 
+LEAGUE_CHAMPIONS: List[Dict[str, Any]] = []
+
 
 def _image_to_list_format(path: str) -> List[int]:
     """
@@ -153,16 +156,85 @@ def _create_the_finals_query_array(highlight: Highlight) -> List[int]:
     return queries
 
 
+def _create_league_of_legends_query_array(highlight: Highlight) -> List[int]:
+    global LEAGUE_CHAMPIONS
+    if not LEAGUE_CHAMPIONS:
+        if not os.path.exists(LEAGUE_IMAGES_PATH):  # pragma: no cover
+            logger.error("The league of legends images do not exists")
+            return []
+
+        for image_path in sorted(os.listdir(LEAGUE_IMAGES_PATH)):
+            LEAGUE_CHAMPIONS.append(
+                {
+                    "image": cv2.imread(os.path.join(LEAGUE_IMAGES_PATH, image_path)),
+                    "name": image_path.split(".")[0],
+                }
+            )
+
+    images = sorted(os.listdir(highlight.images_path))
+    images.sort()
+
+    queries = []
+    yellow_rgb = np.array([54, 216, 213])
+    yellow_threshold = 115
+
+    for j, image_name in enumerate(images):
+        full_image_path = os.path.join(highlight.images_path, image_name)
+        image = cv2.imread(full_image_path, cv2.IMREAD_COLOR)
+
+        is_kill = False
+        kill_spots = []
+
+        regions = {
+            1: image[7:51, 46:48],
+            2: image[69:113, 46:48],
+            3: image[132:176, 46:48],
+            4: image[195:239, 46:48],
+        }
+
+        enemy_region = {
+            1: image[8:49, 84:125],
+            2: image[70:111, 84:125],
+            3: image[133:174, 84:125],
+            4: image[196:237, 84:125],
+        }
+
+        for region_index, region in regions.items():
+            avg_color = np.mean(region.reshape(-1, 3), axis=0)
+            color_distance = np.linalg.norm(avg_color - yellow_rgb)
+
+            if color_distance < yellow_threshold:
+                enemy_image = enemy_region[region_index]
+
+                max_score = -1
+                for champion in LEAGUE_CHAMPIONS:
+                    score = cv2.matchTemplate(
+                        enemy_image, champion["image"], cv2.TM_CCOEFF_NORMED
+                    )
+                    if score > max_score:
+                        max_score = score
+
+                if max_score > 0.75:
+                    is_kill = True
+                    kill_spots.append(region_index)
+
+        if is_kill:
+            queries.append(j)
+    return queries
+
+
 def _get_query_array(
-    neural_network: NeuralNetwork,
+    neural_network: Union[NeuralNetwork, None],
     highlight: Highlight,
     confidence: float,
     game: SupportedGames,
 ) -> List[int]:
     if neural_network:
         return _create_query_array(neural_network, highlight, confidence)
-    if game == SupportedGames.THEFINALS:
+    if game == SupportedGames.THE_FINALS:
         return _create_the_finals_query_array(highlight)
+    if game == SupportedGames.LEAGUE_OF_LEGENDS:
+        return _create_league_of_legends_query_array(highlight)
     raise ValueError(
         f"No neural network for game {game} and no custom query array"
     )  # pragma: no cover
@@ -231,7 +303,7 @@ def _post_process_query_array(
 
 async def extract_segments(
     highlight: Highlight,
-    neural_network: NeuralNetwork,
+    neural_network: Union[NeuralNetwork, None],
     confidence: float,
     framerate: int,
     offset: int,

crispy-api/requirements-dev.txt

@@ -7,3 +7,4 @@ pytest-cov==4.0.0
 flake8
 httpx
 mutagen
+types-requests==2.32.0.20241016

crispy-api/requirements.txt

@@ -21,3 +21,5 @@ scipy==1.8.0
 pydub==0.25.1
 montydb==2.4.0
 easyocr==1.7.1
+aiohttp==3.10.10
+requests==2.32.3

crispy-api/settings.json

@@ -4,9 +4,9 @@
   },
   "clip": {
     "framerate": 8,
-    "second-before": 3,
+    "second-before": 4,
     "second-after": 0.5,
-    "second-between-kills": 1
+    "second-between-kills": 3
   },
   "stretch": false,
   "game": "valorant"