update renderer to have cap status also tickets (#1325)

This commit is contained in:
NotSoToothless
2026-06-14 22:36:06 -07:00
committed by GitHub
parent 4b75ce1533
commit deb4e0fb12
4 changed files with 355 additions and 4 deletions
+168 -2
View File
@@ -32,7 +32,7 @@ import numpy as np
from . import SHARED_DIR
from .utils import REPLAYS_DIR
from PIL import Image, ImageDraw, ImageFilter, ImageFont
from PIL import Image, ImageChops, ImageDraw, ImageFilter, ImageFont
try:
from data_parser import (
@@ -519,6 +519,8 @@ class VideoCtx:
# Kill/damage target icons
kill_target_spr: Sprite | None = None
dmg_target_spr: Sprite | None = None
# Dynamic capture-state fills (per-cap, per-frame); empty for pre-v2 replays
cap_overlays: list = field(default_factory=list)
# ── Map / LevelDef load ────────────────────────────────────────────────────────
@@ -1088,6 +1090,12 @@ def _map_name_candidates(level_path: str, level_def: dict | None,
seen.add(name)
out.append(name)
# The canonical ground map is "<level stem>_tankmap" — this is exactly what the
# web viewer requests (/api/match/minimap/<stem>), so try it first to stay in
# sync with the canvas. customLevelMap points at the full/thumbnail map and
# must not win over the real tankmap for a ground render.
add(f"{stem}_tankmap")
if isinstance(level_def, dict):
tank_custom = level_def.get("customLevelTankMap", "")
if isinstance(tank_custom, str) and tank_custom:
@@ -1328,6 +1336,110 @@ def _draw_capture_areas(img: Image.Image, capture_areas: list[dict],
return out
# ── Capture-state fill (dynamic, per-frame) ────────────────────────────────────
# Colors match the web canvas (RC.WIN / RC.LOSE).
CAP_FILL_WIN = (0, 200, 0)
CAP_FILL_LOSE = (220, 30, 30)
CAP_FILL_ALPHA = 0.5
def _precompute_cap_overlays(capture_areas: list[dict], zones: dict, winner_slot: int,
frame_times: np.ndarray, c0: list[float], c1: list[float],
canvas: int) -> list[dict]:
"""Build per-cap, per-frame fill data from the Spectra zone `cap` timelines.
`cap` sign is the owning team slot (positive == slot 2, negative == slot 1);
a cap is the winner's when that slot == ``winner_slot``. capture_areas is
index-mapped to
zone letters A/B/C…; each overlay carries the diamond outline (for clipping),
center, and per-frame fill fraction + winner/loser flag.
"""
overlays: list[dict] = []
if not capture_areas or not isinstance(zones, dict) or not zones:
return overlays
x0, z0 = float(c0[0]), float(c0[1])
x1, z1 = float(c1[0]), float(c1[1])
xr, zr = x1 - x0, z1 - z0
if xr == 0 or zr == 0:
return overlays
labels = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
for idx, cap in enumerate(capture_areas):
letter = labels[idx] if idx < len(labels) else str(idx + 1)
series = zones.get(letter)
if not series:
continue
ts = np.array([float(p[0]) for p in series], dtype=np.float64)
vs = np.array([float(p[1]) for p in series], dtype=np.float64)
order = np.argsort(ts)
ts, vs = ts[order], vs[order]
# Zero-order hold: a cap holds its last recorded value until the next
# change. Samples are only emitted when the value changes, so linear
# interpolation across the long initial [0,0]→first-change gap would show
# a false early fill before any player has reached the point.
idx = np.clip(np.searchsorted(ts, frame_times, side="right") - 1, 0, len(vs) - 1)
val = vs[idx]
frac = np.clip(np.abs(val) / 100.0, 0.0, 1.0).astype(np.float32)
# cap sign is the owning team slot: negative == slot 1, positive == slot 2
# (verified against unit occupancy in the zone).
owner_slot = np.where(val > 0, 2, 1)
is_winner = (owner_slot == winner_slot)
cx = float(cap.get("x", 0.0))
cz = float(cap.get("z", 0.0))
px, py = _world_to_map_px(cx, cz, x0, z0, xr, zr, canvas)
outline = _cap_outline_points_px(cap, x0, z0, xr, zr, canvas)
rp = 0
if len(outline) < 3:
outline = []
rr = max(0.0, float(cap.get("radius", 0.0)))
rp = max(8, int(round(rr * ((canvas / xr + canvas / zr) * 0.5))))
overlays.append({
"cx": int(px), "cy": int(py), "rp": int(rp),
"outline": outline, "frac": frac, "is_winner": is_winner,
})
return overlays
def _draw_cap_fill_np(buf: np.ndarray, overlay: dict, fi: int) -> None:
"""Alpha-composite a single cap's radial (pie) fill into the frame buffer."""
frac = float(overlay["frac"][fi])
if frac <= 0.01:
return
rgb = CAP_FILL_WIN if bool(overlay["is_winner"][fi]) else CAP_FILL_LOSE
cx, cy = overlay["cx"], overlay["cy"]
outline = overlay["outline"]
if outline:
xs = [p[0] for p in outline]
ys = [p[1] for p in outline]
minx, maxx, miny, maxy = min(xs), max(xs), min(ys), max(ys)
else:
rp = overlay["rp"]
minx, maxx, miny, maxy = cx - rp, cx + rp, cy - rp, cy + rp
minx = max(0, minx)
miny = max(0, miny)
maxx = min(buf.shape[1] - 1, maxx)
maxy = min(buf.shape[0] - 1, maxy)
w, h = maxx - minx, maxy - miny
if w <= 0 or h <= 0:
return
R = max(w, h)
pie = Image.new("L", (w, h), 0)
end = -90.0 + 360.0 * frac
ImageDraw.Draw(pie).pieslice(
[cx - minx - R, cy - miny - R, cx - minx + R, cy - miny + R],
-90.0, end, fill=255)
if outline:
poly = Image.new("L", (w, h), 0)
ImageDraw.Draw(poly).polygon([(px - minx, py - miny) for px, py in outline], fill=255)
pie = ImageChops.multiply(pie, poly)
mask = (np.asarray(pie, dtype=np.float32) / 255.0) * CAP_FILL_ALPHA
if not mask.any():
return
region = buf[miny:maxy, minx:maxx].astype(np.float32)
color = np.array(rgb, dtype=np.float32)
m = mask[:, :, None]
buf[miny:maxy, minx:maxx] = (region * (1.0 - m) + color * m).astype(np.uint8)
def load_map_image(level_path: str, level_def: dict | None,
battle_type: str = "", level_settings_path: str = "",
base_coords: tuple[list[float], list[float]] | None = None,
@@ -2322,6 +2434,11 @@ def render_one_ctx(fi: int, buf: np.ndarray, ctx: VideoCtx,
"""
np.copyto(buf, ctx.bg_arr)
# Dynamic capture-state fills (drawn on the map, beneath players)
if ctx.cap_overlays:
for ov in ctx.cap_overlays:
_draw_cap_fill_np(buf, ov, fi)
cv = ctx.canvas
# Ground trails
draw_all_trails_np(buf, fi, ctx.px_all, ctx.py_all, trail_f, ctx.trail_colors_arr,
@@ -2753,6 +2870,11 @@ def render_gob(
print("not found")
raise ValueError(f"No local tankmap image for level '{level}'")
map_img = _draw_capture_areas(map_img, capture_areas, tc0, tc1)
cap_overlays = _precompute_cap_overlays(
capture_areas, d.get("Zones", {}), _to_int(d.get("WinnerSlot"), 0),
frame_times, tc0, tc1, canvas)
if cap_overlays:
print(f"Cap state : {len(cap_overlays)} zones with live ownership")
print("ok")
bg = make_bg(map_img)
@@ -2900,6 +3022,7 @@ def render_gob(
air_headings=air_headings,
kill_target_spr=kill_target,
dmg_target_spr=dmg_target,
cap_overlays=cap_overlays,
)
shared = (shadow_color, trail_f)
@@ -3133,6 +3256,23 @@ def _convert_ws_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, Any]:
difficulty = str(replay.get("difficulty") or "")
battle_type = mission_mode or difficulty
# Zones (positive cap == team slot 1) and tickets (keyed by slot) are kept in
# their original slot space; readers compare the owning slot to WinnerSlot.
# center/radius are dropped — geometry comes from the mission .blk.
zones_src = replay.get("zones") or {}
tickets_src = replay.get("tickets") or {}
zones_out = {
letter: zdata.get("cap", [])
for letter, zdata in zones_src.items()
if isinstance(zdata, dict)
} if isinstance(zones_src, dict) else {}
tickets_out = tickets_src if isinstance(tickets_src, dict) else {}
team_slot_names = {}
if winner_team:
team_slot_names[str(winner_team)] = winner_tag
if loser_team:
team_slot_names[str(loser_team)] = loser_tag
return {
"SessionID": _to_int(replay.get("_id") or replay.get("id"), 0),
"TeamWon": winner_team,
@@ -3145,6 +3285,10 @@ def _convert_ws_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, Any]:
"Entities": entities_out,
"Kills": kills_out,
"DamageReports": damages_out,
"Zones": zones_out,
"Tickets": tickets_out,
"WinnerSlot": winner_team,
"TeamSlotNames": team_slot_names,
}
@@ -3267,6 +3411,12 @@ def _convert_local_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, An
"Afire": bool(dmg.get("afire", False)),
})
# Zones (letter -> cap timeline, positive == slot 1), tickets (slot -> timeline),
# the winning slot, and slot -> name all stay in slot space for the readers.
zones_out = replay.get("zones") if isinstance(replay.get("zones"), dict) else {}
tickets_out = replay.get("tickets") if isinstance(replay.get("tickets"), dict) else {}
team_slot_names = replay.get("team_slot_names") if isinstance(replay.get("team_slot_names"), dict) else {}
return {
"SessionID": _to_int(replay.get("session_id_dec") or replay.get("session_id_hex"), 0),
"TeamWon": winner_team,
@@ -3279,6 +3429,10 @@ def _convert_local_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, An
"Entities": entities_out,
"Kills": kills_out,
"DamageReports": damages_out,
"Zones": zones_out,
"Tickets": tickets_out,
"WinnerSlot": _to_int(replay.get("winner_team_slot"), 0),
"TeamSlotNames": team_slot_names,
}
@@ -3466,7 +3620,7 @@ def export_replay_json(replay_path: Path) -> dict:
"name": p.get("Name", ""),
"team": p.get("Team", 0),
}
clan = p.get("ClanTag", "")
clan = p.get("ClanTag") or p.get("Clan") or ""
if clan:
player["clan"] = clan
players_out.append(player)
@@ -3548,6 +3702,14 @@ def export_replay_json(replay_path: Path) -> dict:
"offendedId": dm.get("OffendedID", 0),
})
# Capture-zone ownership + ticket timelines (Spectra v2+); empty for older
# replays. Kept in slot space: cap sign / ticket keys / team names are all
# keyed by team slot, and `winnerSlot` says which slot won.
capture_state = d.get("Zones") if isinstance(d.get("Zones"), dict) else {}
tickets_series = d.get("Tickets") if isinstance(d.get("Tickets"), dict) else {}
team_names = d.get("TeamSlotNames") if isinstance(d.get("TeamSlotNames"), dict) else {}
winner_slot = d.get("WinnerSlot", 0)
out = {
"teamWon": team_won,
"mission": {
@@ -3558,6 +3720,10 @@ def export_replay_json(replay_path: Path) -> dict:
},
"levelCoords": level_coords,
"captureAreas": capture_areas,
"captureState": capture_state,
"tickets": tickets_series,
"teamNames": team_names,
"winnerSlot": winner_slot,
"players": players_out,
"entities": entities_out,
"kills": kills_out,
+34 -2
View File
@@ -96,8 +96,8 @@ def replay_data_path(session_id: str | int) -> Path:
# When enabled, the unmodified Spectra payload is stored next to the transformed
# replay (as RAW_replay_data.json.gz) so it can be pulled and re-processed later.
# Toggle off by setting SRE_STORE_RAW_REPLAY=0 in the environment.
STORE_RAW_REPLAY = os.getenv("SRE_STORE_RAW_REPLAY", "1").strip().lower() not in (
"0", "false", "no", "off", "",
STORE_RAW_REPLAY = os.getenv("SRE_STORE_RAW_REPLAY", "0").strip().lower() in (
"1", "true", "yes", "on",
)
@@ -1273,6 +1273,10 @@ def transform_to_local_format(api_data: Dict[str, Any]) -> Optional[Dict[str, An
winner_players: List[Dict[str, Any]] = []
loser_players: List[Dict[str, Any]] = []
# Original Spectra team slots of the winner/loser, captured from the same
# tag match below — used to normalise zones/tickets into winner-first space.
winner_team_slot: Optional[str] = None
loser_team_slot: Optional[str] = None
for uid_str, pdata in players_dict.items():
try:
@@ -1307,8 +1311,12 @@ def transform_to_local_format(api_data: Dict[str, Any]) -> Optional[Dict[str, An
# Assign to winner or loser by comparing tag
if tag == winner_winged:
winner_players.append(player_entry)
if winner_team_slot is None:
winner_team_slot = str(pdata.get("team"))
elif tag == loser_winged:
loser_players.append(player_entry)
if loser_team_slot is None:
loser_team_slot = str(pdata.get("team"))
except (ValueError, TypeError) as e:
logging.warning(f"Skipping bad player UID {uid_str}: {e}")
continue
@@ -1469,6 +1477,26 @@ def transform_to_local_format(api_data: Dict[str, Any]) -> Optional[Dict[str, An
except (ValueError, TypeError):
session_id_hex = ""
# Capture-zone ownership + ticket timelines (Spectra v2+), kept in their
# original team-slot space: cap sign is the owning slot (positive == slot
# "2", negative == slot "1"), and tickets stay keyed by slot ("1"/"2"). Readers decide
# win/loss by comparing the owning slot to `winner_team_slot` — no sign
# flipping or re-keying. center/radius are dropped (geometry comes from
# the mission .blk in resolve_capture_areas()).
raw_zones = replay.get("zones") or {}
raw_tickets = replay.get("tickets") or {}
zones = {
letter: zdata.get("cap", [])
for letter, zdata in raw_zones.items()
if isinstance(zdata, dict)
} if isinstance(raw_zones, dict) else {}
tickets = raw_tickets if isinstance(raw_tickets, dict) else {}
team_slot_names = {}
if winner_team_slot:
team_slot_names[winner_team_slot] = winner_squadron
if loser_team_slot:
team_slot_names[loser_team_slot] = loser_squadron
return {
"winning_team_squadron": winner_squadron,
"losing_team_squadron": loser_squadron,
@@ -1509,6 +1537,10 @@ def transform_to_local_format(api_data: Dict[str, Any]) -> Optional[Dict[str, An
"mission_path": replay.get("mission_path"),
"difficulty": replay.get("difficulty"),
"type": replay.get("type", ""),
"zones": zones,
"tickets": tickets,
"winner_team_slot": winner_team_slot,
"team_slot_names": team_slot_names,
}
except Exception as e: