Lesson 20: Lag Compensation

History entries are (tick, (x, y)) pairs, so history[0][0] is the oldest tick and history[-1][1] the newest position. In the ray test, (center - origin).dot(direction) is how far along the ray the target's center lies; clamp it before you use it.

If your instructor hands you the lab file, you will see a few extra lines marked lab runtime near the top and and frame_budget() in the loop. They let the instructor's checker run the program automatically for a fixed number of frames; when you run it yourself they do nothing.

"""Rewind and Hit: Advanced Lesson 20 practice exercise (solution).

A hitscan shooter (you, bottom center) fires at a bot that only the server moves.
Like every client, you see the bot a little in the past: the blue bot is your
interpolated view. The server keeps one second of history and, when your shot
arrives, rewinds the bot to the tick you were looking at before it tests the hit.
  * red outline  = where the bot is on the server right now
  * green / gray = where the server rewound it for your last shot (hit / miss)
Click to fire at the mouse, SPACE to fire at the bot you see, L to change latency.
The HUD compares the verdict with rewind against the verdict without it.
"""
from collections import deque
from dataclasses import dataclass

import pygame


WIDTH, HEIGHT = 800, 450
TICK_RATE = 60                      # server ticks per second
TICK_DT = 1 / TICK_RATE
INTERP_TICKS = 6                    # clients draw others 6 ticks (0.1 s) in the past
MAX_REWIND_TICKS = 18               # the server rewinds at most 0.3 s
HISTORY_TICKS = TICK_RATE           # keep one second of the past
LATENCIES = [0.03, 0.05, 0.12]      # one-way delays to cycle through, seconds
BOT_RADIUS = 20
BOT_SPEED = 260.0                   # px/s
SHOOTER = (WIDTH / 2, HEIGHT - 30)
RANGE = 900.0                       # px a hitscan shot can reach


@dataclass(frozen=True)
class Shot:
    origin: tuple
    aim: tuple                      # the point the player clicked on
    render_tick: float              # the tick the client was DRAWING when it fired


class LaggyLink:
    """One direction of a connection: in order, nothing lost, `delay` seconds late."""

    def __init__(self, delay):
        self.delay = delay
        self.queue = deque()

    def send(self, now, message):
        self.queue.append((now + self.delay, message))

    def receive(self, now):
        out = []
        while self.queue and self.queue[0][0] <= now:
            out.append(self.queue.popleft()[1])
        return out


def state_at(history, tick):
    """The bot's position at a fractional tick, blended from the history.

    history holds (tick, (x, y)) pairs, oldest first. Returns None when the tick is
    older than anything we kept: the server can't know, so the shot is refused.
    """
    if not history or tick < history[0][0]:
        return None
    if tick >= history[-1][0]:
        return pygame.Vector2(history[-1][1])      # newest we have (never the oldest!)
    for (ta, pa), (tb, pb) in zip(history, list(history)[1:]):
        if ta <= tick <= tb:
            span = tb - ta
            t = (tick - ta) / span if span else 0.0  # guard: two entries for one tick
            return pygame.Vector2(pa).lerp(pb, t)
    return pygame.Vector2(history[-1][1])


def ray_hits_circle(origin, aim, center, radius, max_range=RANGE):
    """Hitscan test: does the ray from origin toward aim pass within radius of center?"""
    origin, center = pygame.Vector2(origin), pygame.Vector2(center)
    direction = pygame.Vector2(aim) - origin
    if direction.length_squared() == 0:
        return False
    direction = direction.normalize()
    along = max(0.0, min(max_range, (center - origin).dot(direction)))
    closest = origin + direction * along              # nearest point of the ray to the center
    return closest.distance_squared_to(center) <= radius * radius


class Server:
    def __init__(self):
        self.tick = 0
        self.bot = pygame.Vector2(120, 150)
        self.velocity = pygame.Vector2(BOT_SPEED, 0)
        self.history = deque(maxlen=HISTORY_TICKS)
        self.history.append((self.tick, tuple(self.bot)))

    def step(self):
        self.tick += 1
        self.bot += self.velocity * TICK_DT
        if self.bot.x > WIDTH - 60:
            self.bot.x, self.velocity.x = WIDTH - 60, -BOT_SPEED
        elif self.bot.x < 60:
            self.bot.x, self.velocity.x = 60, BOT_SPEED
        self.history.append((self.tick, tuple(self.bot)))     # a tuple: a copy, not a reference
        return self.tick, tuple(self.bot)

    def judge(self, shot):
        """Return (verdict, rewound position). Verdicts: 'hit', 'miss', 'too old', 'future'."""
        age = self.tick - shot.render_tick
        if age < 0:
            return "future", None
        if age > MAX_REWIND_TICKS:
            return "too old", None
        past = state_at(self.history, shot.render_tick)
        if past is None:
            return "too old", None
        if ray_hits_circle(shot.origin, shot.aim, past, BOT_RADIUS):
            return "hit", past
        return "miss", past

    def judge_without_rewind(self, shot):
        return "hit" if ray_hits_circle(shot.origin, shot.aim, self.bot, BOT_RADIUS) else "miss"


class Client:
    def __init__(self):
        self.snapshots = deque(maxlen=64)             # (tick, (x, y))
        self.clock = None                             # estimated newest snapshot tick, float

    def on_snapshot(self, snap):
        self.snapshots.append(snap)
        if self.clock is None or abs(self.clock - snap[0]) > INTERP_TICKS:
            self.clock = float(snap[0])

    def advance_clock(self, dt):
        if self.clock is not None:
            self.clock += dt * TICK_RATE

    def render_tick(self):
        """The tick we draw: INTERP_TICKS behind our clock, but only ticks we have received."""
        if self.clock is None:
            return None
        newest, oldest = self.snapshots[-1][0], self.snapshots[0][0]
        return max(oldest, min(newest, self.clock - INTERP_TICKS))

    def bot_view(self):
        rt = self.render_tick()
        return None if rt is None else state_at(self.snapshots, rt)


def main():
    pygame.init()
    screen = pygame.display.set_mode((WIDTH, HEIGHT))
    pygame.display.set_caption("Rewind and Hit")
    clock = pygame.time.Clock()
    font = pygame.font.Font(None, 24)

    server, client = Server(), Client()
    latency_index = 1
    up = LaggyLink(LATENCIES[latency_index])
    down = LaggyLink(LATENCIES[latency_index])
    now, tick_timer = 0.0, 0.0
    counts = {"shots": 0, "rewind hits": 0, "plain hits": 0, "refused": 0}
    last = None                                        # (verdict, rewound position)
    tracer = None                                      # (aim point, seconds left)

    running = True
    while running:
        dt = min(clock.tick(60) / 1000, 0.1)
        now += dt
        for event in pygame.event.get():
            aim = None
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.MOUSEBUTTONDOWN and event.button == 1:
                aim = event.pos
            elif event.type == pygame.KEYDOWN and event.key == pygame.K_SPACE:
                view = client.bot_view()
                aim = tuple(view) if view is not None else None
            elif event.type == pygame.KEYDOWN and event.key == pygame.K_l:
                latency_index = (latency_index + 1) % len(LATENCIES)
                up.delay = down.delay = LATENCIES[latency_index]
            if aim is not None and client.render_tick() is not None:
                up.send(now, Shot(SHOOTER, tuple(aim), client.render_tick()))
                tracer = (aim, 0.25)

        tick_timer += dt
        while tick_timer >= TICK_DT:
            tick_timer -= TICK_DT
            down.send(now, server.step())
            for shot in up.receive(now):
                verdict, past = server.judge(shot)
                counts["shots"] += 1
                if verdict == "hit":
                    counts["rewind hits"] += 1
                elif verdict in ("too old", "future"):
                    counts["refused"] += 1
                if server.judge_without_rewind(shot) == "hit":
                    counts["plain hits"] += 1
                last = (verdict, past)

        for snap in down.receive(now):
            client.on_snapshot(snap)
        client.advance_clock(dt)
        if tracer is not None:
            tracer = (tracer[0], tracer[1] - dt) if tracer[1] > dt else None

        screen.fill((16, 20, 32))
        pygame.draw.circle(screen, (240, 90, 90), server.bot, BOT_RADIUS, 2)
        view = client.bot_view()
        if view is not None:
            pygame.draw.circle(screen, (90, 160, 250), view, BOT_RADIUS)
        if last is not None and last[1] is not None:
            color = (90, 230, 120) if last[0] == "hit" else (150, 150, 160)
            pygame.draw.circle(screen, color, last[1], BOT_RADIUS, 3)
        if tracer is not None:
            pygame.draw.line(screen, (255, 230, 120), SHOOTER, tracer[0], 2)
        pygame.draw.circle(screen, (230, 230, 240), SHOOTER, 10)
        one_way = LATENCIES[latency_index]
        rewind_ticks = 2 * one_way * TICK_RATE + INTERP_TICKS
        lines = [
            f"One-way {one_way * 1000:.0f} ms, round trip {2 * one_way * 1000:.0f} ms; "
            f"rewind about {rewind_ticks:.0f} ticks (limit {MAX_REWIND_TICKS})   [L]",
            f"Shots {counts['shots']}: with rewind {counts['rewind hits']} hits, "
            f"without rewind {counts['plain hits']} hits, refused {counts['refused']}",
            f"Last verdict: {last[0] if last else '-'}   (SPACE: fire at what you see, click: fire at mouse)",
        ]
        for i, line in enumerate(lines):
            screen.blit(font.render(line, True, (225, 228, 235)), (12, 10 + i * 22))
        pygame.display.flip()

    pygame.quit()
    print(f"Shots judged: {counts['shots']}")
    print(f"With rewind: {counts['rewind hits']} hits, refused {counts['refused']}")
    print(f"Without rewind: {counts['plain hits']} hits")


if __name__ == "__main__":
    main()