feat: heatmap + evo + decay
This commit is contained in:
@@ -0,0 +1,198 @@
|
||||
"""Renders one event's pixel placement history as a heatmap.
|
||||
|
||||
Same source as gif.py (the durable PixelPlacement history), but instead of
|
||||
replaying colors, counts how many times each pixel was placed and maps that
|
||||
count to a white -> red gradient, so an admin can see
|
||||
where the action was - either as a single PNG of the final counts, as an
|
||||
animated GIF of the counts building up over time, or as an animated GIF
|
||||
where the heat decays so only recent activity stays red.
|
||||
"""
|
||||
|
||||
import math
|
||||
from io import BytesIO
|
||||
|
||||
from django.db.models import Count
|
||||
from PIL import Image
|
||||
|
||||
from events.models import Event
|
||||
|
||||
from .gif import FINAL_FRAME_HOLD_MS, FRAME_DURATION_MS, MAX_FRAMES
|
||||
from .models import PixelPlacement, PlaceSettings
|
||||
|
||||
SCALE = 8
|
||||
|
||||
# Gradient stops, from "never placed" to "most placed"
|
||||
_GRADIENT = [
|
||||
(255, 255, 255),
|
||||
(193, 1, 2),
|
||||
]
|
||||
|
||||
|
||||
def build_heatmap_png(event: Event, usernames: list[str] | None = None) -> bytes:
|
||||
"""Renders `event`'s placement counts per pixel as a PNG heatmap,
|
||||
upscaled for visibility. Returns PNG bytes, even if there are zero
|
||||
placements (all white).
|
||||
|
||||
If `usernames` is given, only those players' placements are counted
|
||||
(combined on the same board).
|
||||
|
||||
Counts are log-scaled, otherwise a few heavily fought-over pixels make
|
||||
everything else look white. Placements outside the *current* grid
|
||||
dimensions are skipped, like in gif.py."""
|
||||
place_settings = PlaceSettings.current(event)
|
||||
width, height = place_settings.width, place_settings.height
|
||||
|
||||
placements = _placements(event, usernames)
|
||||
counts = placements.values("x", "y").annotate(count=Count("id")).order_by()
|
||||
|
||||
grid = [0] * (width * height)
|
||||
for row in counts:
|
||||
if row["x"] >= width or row["y"] >= height:
|
||||
continue
|
||||
grid[row["y"] * width + row["x"]] = row["count"]
|
||||
|
||||
colors = _color_table(max(grid, default=0))
|
||||
buffer = BytesIO()
|
||||
_render_frame(grid, width, height, colors).save(buffer, format="PNG")
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def build_heatmap_evolution_gif(
|
||||
event: Event, usernames: list[str] | None = None
|
||||
) -> bytes:
|
||||
"""Same as build_heatmap_png, but as an animated GIF where the counts
|
||||
accumulate placement by placement (sampled down to MAX_FRAMES, like
|
||||
gif.py). Returns GIF bytes, even if there are zero placements.
|
||||
|
||||
Every frame uses the *final* max count for its color scale, so a pixel's
|
||||
color only ever heats up - otherwise early frames would look as hot as
|
||||
the end."""
|
||||
place_settings = PlaceSettings.current(event)
|
||||
width, height = place_settings.width, place_settings.height
|
||||
|
||||
placements = [
|
||||
(x, y)
|
||||
for x, y in _placements(event, usernames)
|
||||
.order_by("placed_at")
|
||||
.values_list("x", "y")
|
||||
if x < width and y < height
|
||||
]
|
||||
|
||||
final_grid = [0] * (width * height)
|
||||
for x, y in placements:
|
||||
final_grid[y * width + x] += 1
|
||||
colors = _color_table(max(final_grid, default=0))
|
||||
|
||||
grid = [0] * (width * height)
|
||||
frames: list[Image.Image] = []
|
||||
|
||||
step = max(1, len(placements) // MAX_FRAMES)
|
||||
for i, (x, y) in enumerate(placements):
|
||||
grid[y * width + x] += 1
|
||||
if i % step == 0 or i == len(placements) - 1:
|
||||
frames.append(_render_frame(grid, width, height, colors))
|
||||
|
||||
if not frames:
|
||||
frames.append(_render_frame(grid, width, height, colors))
|
||||
|
||||
return _save_gif(frames)
|
||||
|
||||
|
||||
def build_heatmap_decay_gif(
|
||||
event: Event, usernames: list[str] | None = None, half_life: int | None = None
|
||||
) -> bytes:
|
||||
"""Like build_heatmap_evolution_gif, but every pixel's heat decays
|
||||
exponentially, so a pixel nobody touches anymore fades back to white
|
||||
and the GIF shows where the activity is *right now*.
|
||||
|
||||
`half_life` is in placements (after that many more placements anywhere
|
||||
on the board, a pixel's heat is halved). Defaults to 10 frames' worth.
|
||||
|
||||
Every frame shares the same color scale (the hottest pixel across the
|
||||
whole replay), so a quiet period actually looks quiet."""
|
||||
place_settings = PlaceSettings.current(event)
|
||||
width, height = place_settings.width, place_settings.height
|
||||
|
||||
placements = [
|
||||
(x, y)
|
||||
for x, y in _placements(event, usernames)
|
||||
.order_by("placed_at")
|
||||
.values_list("x", "y")
|
||||
if x < width and y < height
|
||||
]
|
||||
|
||||
step = max(1, len(placements) // MAX_FRAMES)
|
||||
decay = 0.5 ** (1 / max(1, half_life or step * 10))
|
||||
|
||||
# Decay is applied lazily: each pixel stores its heat as of the
|
||||
# placement index it was last touched at, so a placement is O(1)
|
||||
# instead of decaying the whole grid every time.
|
||||
heat = [0.0] * (width * height)
|
||||
touched_at = [0] * (width * height)
|
||||
snapshots: list[list[float]] = []
|
||||
for i, (x, y) in enumerate(placements):
|
||||
index = y * width + x
|
||||
heat[index] = heat[index] * decay ** (i - touched_at[index]) + 1
|
||||
touched_at[index] = i
|
||||
if i % step == 0 or i == len(placements) - 1:
|
||||
snapshots.append([h * decay ** (i - t) for h, t in zip(heat, touched_at)])
|
||||
|
||||
if not snapshots:
|
||||
snapshots.append(heat)
|
||||
|
||||
max_log = math.log1p(max(max(snapshot) for snapshot in snapshots)) or 1
|
||||
colors = [_gradient(i / 255) for i in range(256)]
|
||||
frames = []
|
||||
for snapshot in snapshots:
|
||||
image = Image.new("RGB", (width, height))
|
||||
image.putdata([colors[round(math.log1p(h) / max_log * 255)] for h in snapshot])
|
||||
frames.append(image.resize((width * SCALE, height * SCALE), Image.NEAREST))
|
||||
|
||||
return _save_gif(frames)
|
||||
|
||||
|
||||
def _save_gif(frames: list[Image.Image]) -> bytes:
|
||||
buffer = BytesIO()
|
||||
durations = [FRAME_DURATION_MS] * len(frames)
|
||||
durations[-1] = FINAL_FRAME_HOLD_MS
|
||||
frames[0].save(
|
||||
buffer,
|
||||
format="GIF",
|
||||
save_all=True,
|
||||
append_images=frames[1:],
|
||||
duration=durations,
|
||||
loop=0,
|
||||
disposal=2,
|
||||
)
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def _placements(event: Event, usernames: list[str] | None):
|
||||
placements = PixelPlacement.objects.filter(event=event)
|
||||
if usernames:
|
||||
placements = placements.filter(player__username__in=usernames)
|
||||
return placements
|
||||
|
||||
|
||||
def _color_table(max_count: int) -> list[tuple[int, int, int]]:
|
||||
"""Color for every count in [0, max_count], log-scaled."""
|
||||
max_log = math.log1p(max_count) or 1
|
||||
return [_gradient(math.log1p(count) / max_log) for count in range(max_count + 1)]
|
||||
|
||||
|
||||
def _render_frame(
|
||||
grid: list[int], width: int, height: int, colors: list[tuple[int, int, int]]
|
||||
) -> Image.Image:
|
||||
image = Image.new("RGB", (width, height))
|
||||
image.putdata([colors[count] for count in grid])
|
||||
return image.resize((width * SCALE, height * SCALE), Image.NEAREST)
|
||||
|
||||
|
||||
def _gradient(t: float) -> tuple[int, int, int]:
|
||||
"""Linearly interpolates `_GRADIENT` at `t` in [0, 1]."""
|
||||
position = t * (len(_GRADIENT) - 1)
|
||||
index = min(int(position), len(_GRADIENT) - 2)
|
||||
frac = position - index
|
||||
start, end = _GRADIENT[index], _GRADIENT[index + 1]
|
||||
r, g, b = (round(a + (z - a) * frac) for a, z in zip(start, end))
|
||||
return r, g, b
|
||||
Reference in New Issue
Block a user