gacha_renderer: font caching, threaded bg/png, spark_target; gacha: file i/o out of lock; paths from pkg root

This commit is contained in:
esef committed 2026-08-03 20:08:27 +03:00
1 parent 30efb9105f
commit 3ef57f85fe
4 files changed
+51 -24

No files matched your search

+5 -3
View File
@@ -4,9 +4,11 @@
from pathlib import Path from pathlib import Path
# ── Paths ─────────────────────────────────────────────────────────────────── # ── Paths ───────────────────────────────────────────────────────────────────
BUNDLED_DB = Path(__file__).parent / "ba_students.json" _PKG_DIR = Path(__file__).parent
BANNER_FILE = Path("data/ba_banner_state.json") _DATA_DIR = _PKG_DIR.parent / "data"
ASSETS_DIR = Path(__file__).parent / "assets" BUNDLED_DB = _PKG_DIR / "ba_students.json"
BANNER_FILE = _DATA_DIR / "ba_banner_state.json"
ASSETS_DIR = _PKG_DIR / "assets"
BG_PATH = ASSETS_DIR / "gacha_bg.png" BG_PATH = ASSETS_DIR / "gacha_bg.png"
GACHA_ANIM_PATH = ASSETS_DIR / "Arona_Blue_Gacha_Animation.gif" GACHA_ANIM_PATH = ASSETS_DIR / "Arona_Blue_Gacha_Animation.gif"
GACHA_ANIM_DURATION = 3.5 # 195 frames × 100 ms (used to be 19.5 will be changed later) GACHA_ANIM_DURATION = 3.5 # 195 frames × 100 ms (used to be 19.5 will be changed later)
+2 -1
View File
@@ -11,7 +11,8 @@ from .constants import ELIGMA_YIELD
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
DB_PATH = Path("data/ba_gacha.db") _PKG_DIR = Path(__file__).parent
DB_PATH = _PKG_DIR.parent / "data" / "ba_gacha.db"
def _connect() -> sqlite3.Connection: def _connect() -> sqlite3.Connection:
+17 -6
View File
@@ -25,21 +25,32 @@ log = logging.getLogger(__name__)
_banner_lock = asyncio.Lock() _banner_lock = asyncio.Lock()
async def _load_banner_state() -> dict: def _read_banner_file() -> dict:
"""Which banner each user has picked.""" """Read banner state from disk (runs in thread)."""
async with _banner_lock:
if BANNER_FILE.exists(): if BANNER_FILE.exists():
with open(BANNER_FILE, "r") as f: with open(BANNER_FILE, "r") as f:
return json.load(f) return json.load(f)
return {} return {}
async def _save_banner_state(state: dict) -> None: def _write_banner_file(state: dict) -> None:
"""Write banner state to disk (runs in thread)."""
BANNER_FILE.parent.mkdir(parents=True, exist_ok=True) BANNER_FILE.parent.mkdir(parents=True, exist_ok=True)
async with _banner_lock, open(BANNER_FILE, "w") as f: with open(BANNER_FILE, "w") as f:
json.dump(state, f, indent=2) json.dump(state, f, indent=2)
async def _load_banner_state() -> dict:
"""Which banner each user has picked."""
async with _banner_lock:
return await asyncio.to_thread(_read_banner_file)
async def _save_banner_state(state: dict) -> None:
async with _banner_lock:
await asyncio.to_thread(_write_banner_file, state)
async def _get_banner_id(user_id: int) -> Optional[str]: async def _get_banner_id(user_id: int) -> Optional[str]:
state = await _load_banner_state() state = await _load_banner_state()
return state.get(str(user_id)) return state.get(str(user_id))
@@ -63,7 +74,7 @@ class BlueArchiveGacha(commands.Cog):
async def cog_load(self) -> None: async def cog_load(self) -> None:
"""Load student DB, init SQLite, and fetch banners on startup.""" """Load student DB, init SQLite, and fetch banners on startup."""
db.load() await asyncio.to_thread(db.load)
await gacha_db.init_db() await gacha_db.init_db()
await self._refresh_banners() await self._refresh_banners()
await db.merge_ennead(self._banner_cache) await db.merge_ennead(self._banner_cache)
+27 -14
View File
@@ -2,8 +2,10 @@
"""Pillow-based compositor: renders an in-game-style gacha recruitment result screen.""" """Pillow-based compositor: renders an in-game-style gacha recruitment result screen."""
###################################################################### ######################################################################
import asyncio import asyncio
import functools
import io import io
import logging import logging
import math
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
@@ -19,7 +21,7 @@ from .constants import (
HEADER_Y, FOOTER_Y, HEADER_Y, FOOTER_Y,
STAR_BANNER_H, NAME_STRIP_H, PORTRAIT_TOP, PORTRAIT_H, STAR_BANNER_H, NAME_STRIP_H, PORTRAIT_TOP, PORTRAIT_H,
RARITY_COLOR, RARITY_BG, RARITY_COLOR, RARITY_BG,
BG_PATH, BG_PATH, SPARK_TARGET,
FONT_PATHS, FONT_PATHS,
) )
@@ -27,8 +29,9 @@ log = logging.getLogger(__name__)
# ── Font helpers ─────────────────────────────────────────────────────────── # ── Font helpers ───────────────────────────────────────────────────────────
@functools.lru_cache(maxsize=16)
def _get_font(size: int, bold: bool = False) -> ImageFont.FreeTypeFont: def _get_font(size: int, bold: bool = False) -> ImageFont.FreeTypeFont:
"""Try to load a nice font; fall back to default.""" """Try to load a nice font; fall back to default. Cached per (size, bold)."""
for fp in FONT_PATHS: for fp in FONT_PATHS:
if Path(fp).exists(): if Path(fp).exists():
try: try:
@@ -86,7 +89,6 @@ def _star_polygon(cx: int, cy: int, outer_r: int, inner_r: int) -> list[tuple[in
Uses 10 alternating outer/inner radius points, rotated by pi/5 each step. Uses 10 alternating outer/inner radius points, rotated by pi/5 each step.
""" """
import math
points = [] points = []
for i in range(10): for i in range(10):
angle = math.pi / 2 + i * math.pi / 5 # start from top angle = math.pi / 2 + i * math.pi / 5 # start from top
@@ -118,6 +120,23 @@ def _draw_star_icons(
draw.polygon(pts, fill=color) draw.polygon(pts, fill=color)
# ── Sync render helpers (run in thread) ────────────────────────────────────
def _load_background() -> Image.Image:
"""Load or create the render canvas background."""
if BG_PATH.exists():
return Image.open(BG_PATH).convert("RGBA")
return Image.new("RGBA", (CANVAS_W, CANVAS_H), (180, 226, 245))
def _encode_png(canvas: Image.Image) -> io.BytesIO:
"""Encode a Pillow image to optimized PNG bytes."""
output = io.BytesIO()
canvas.convert("RGB").save(output, format="PNG", optimize=True)
output.seek(0)
return output
# ── Main render ──────────────────────────────────────────────────────────── # ── Main render ────────────────────────────────────────────────────────────
async def render_pull( async def render_pull(
@@ -135,11 +154,8 @@ async def render_pull(
Returns: Returns:
BytesIO containing the rendered PNG image. BytesIO containing the rendered PNG image.
""" """
# Load background # Load background (file I/O off the event loop)
if BG_PATH.exists(): canvas = await asyncio.to_thread(_load_background)
canvas = Image.open(BG_PATH).convert("RGBA")
else:
canvas = Image.new("RGBA", (CANVAS_W, CANVAS_H), (180, 226, 245))
draw = ImageDraw.Draw(canvas) draw = ImageDraw.Draw(canvas)
# ── Header ────────────────────────────────────────────────────────── # ── Header ──────────────────────────────────────────────────────────
@@ -167,11 +183,8 @@ async def render_pull(
cy = GRID_Y + row * (CARD_H + GAP_Y) cy = GRID_Y + row * (CARD_H + GAP_Y)
_draw_card(canvas, draw, cx, cy, student, portrait) _draw_card(canvas, draw, cx, cy, student, portrait)
# ── Encode ────────────────────────────────────────────────────────── # ── Encode (CPU-bound — off the event loop) ─────────────────────────
output = io.BytesIO() return await asyncio.to_thread(_encode_png, canvas)
canvas.convert("RGB").save(output, format="PNG", optimize=True)
output.seek(0)
return output
def _draw_header(draw: ImageDraw.Draw, banner_name: str) -> None: def _draw_header(draw: ImageDraw.Draw, banner_name: str) -> None:
@@ -202,7 +215,7 @@ def _draw_header(draw: ImageDraw.Draw, banner_name: str) -> None:
def _draw_footer(draw: ImageDraw.Draw, spark_count: int) -> None: def _draw_footer(draw: ImageDraw.Draw, spark_count: int) -> None:
"""Draw spark counter at the bottom.""" """Draw spark counter at the bottom."""
text = f"Recruitment Points: {spark_count} / 200" text = f"Recruitment Points: {spark_count} / {SPARK_TARGET}"
tw = draw.textlength(text, font=FOOTER_FONT) tw = draw.textlength(text, font=FOOTER_FONT)
draw.text( draw.text(
((CANVAS_W - tw) / 2 + 1, FOOTER_Y + 1), ((CANVAS_W - tw) / 2 + 1, FOOTER_Y + 1),