"""Headless Zero-G Constant-Speed simulation engine with exact mathematical rigid-body physics.""" from __future__ import annotations import math from raceforge.schemas.spec import RacerID, RaceSpec from raceforge.schemas.timeline import ( EventKind, FrameRacerTransform, SimEvent, SimOutcome, Timeline, ) from raceforge.track.grid import CellType, OccupancyGrid, WallBox from raceforge.track.solver import GeodesicSolver from squarecore.track.hazards import compute_hydraulic_press_y def compute_target_speed(spec: RaceSpec, frame: int, fps: int = 60) -> float: """Evaluate deterministic speed escalation curve as a pure function of frame index.""" t = frame / float(fps) ramp = spec.physics.speed_ramp base = spec.physics.base_speed if t <= ramp.ramp_start_s: return base if t >= ramp.ramp_end_s: return base * ramp.end_mult # Smoothstep interpolation xi = (t - ramp.ramp_start_s) / (ramp.ramp_end_s - ramp.ramp_start_s) smooth = xi * xi * (3.0 - 2.0 * xi) mult = ramp.start_mult + (ramp.end_mult - ramp.start_mult) * smooth return base * mult class SimulationEngine: """Pure-simulation headless runner producing a validated Timeline using exact mathematical physics.""" def __init__( self, spec: RaceSpec, grid: OccupancyGrid, solver: GeodesicSolver, fps: int = 60, max_duration_s: float = 75.0, countdown_s: float = 0.0, ) -> None: self.spec = spec self.grid = grid self.solver = solver self.fps = fps self.max_frames = int(max_duration_s * fps) self.countdown_s = getattr(spec, "countdown_s", countdown_s) self.countdown_frames = int(self.countdown_s * fps) self.goal_tile = spec.track.goal_tile self.goal_cx = (self.goal_tile[0] + self.goal_tile[2]) / 2.0 self.goal_cy = (self.goal_tile[1] + self.goal_tile[3]) / 2.0 self.sz = spec.physics.racer_size self.hs = self.sz / 2.0 self.total_r = self.hs # Breakable Cells setup self.breakable_cells: dict[tuple[int, int], int] = {} track = getattr(spec, "track", None) if track is not None: try: for k, v in getattr(track, "breakable_cells", {}).items(): parts = str(k).split(",") self.breakable_cells[(int(parts[0]), int(parts[1]))] = int(v) except Exception: pass # Individual Solid WallBox objects (exact 1-to-1 match with visual blocks drawn on screen) self.wall_boxes = grid.extract_wall_boxes(breakable_cells=set(self.breakable_cells.keys())) # Pinball Bumpers setup self.bumper_radius = 24.0 self.bumper_positions: list[tuple[float, float, tuple[int, int]]] = [] if track is not None: try: for (br, bc) in list(getattr(track, "bumper_cells", []) or []): cx, cy = grid.cell_center(int(bc), int(br)) self.bumper_positions.append((cx, cy, (int(br), int(bc)))) except Exception: pass self.bumper_impulse = float(getattr(spec.physics, "bumper_impulse", 1100.0)) self.bumper_hits = 0 self.blocks_shattered = 0 self.last_stomp_cycle: int = -1 # Spatial hash grid for 100x collision acceleration and zero phantom bounces self.spatial_boxes: dict[tuple[int, int], list[WallBox]] = {} cell_sz = grid.cell_size off_x = grid.offset_x off_y = grid.offset_y for wb in self.wall_boxes: c_min = max(0, min(grid.cols - 1, int((wb.left - self.hs - off_x) / cell_sz))) c_max = max(0, min(grid.cols - 1, int((wb.right + self.hs - off_x) / cell_sz))) r_min = max(0, min(grid.rows - 1, int((wb.top - self.hs - off_y) / cell_sz))) r_max = max(0, min(grid.rows - 1, int((wb.bottom + self.hs - off_y) / cell_sz))) for c in range(c_min, c_max + 1): for r in range(r_min, r_max + 1): self.spatial_boxes.setdefault((c, r), []).append(wb) # Exact arena boundaries matching Cairo frame (outer line width 12 at 54.0) self.arena_min_x = 60.0 + self.hs self.arena_max_x = 1020.0 - self.hs self.arena_min_y = 60.0 + self.hs self.arena_max_y = 1860.0 - self.hs # Descending Rigid Flood Wall (45s Moving Solid Barrier) self.flood_wall_enabled = bool(getattr(spec, "flood_wall_enabled", True)) self.flood_wall_duration_s = float(getattr(spec, "flood_wall_duration_s", 45.0)) self.flood_wall_y_start = self.arena_min_y - self.hs # 60.0 px self.flood_wall_y_finish = self.goal_cy self.flood_wall_vy = (self.flood_wall_y_finish - self.flood_wall_y_start) / max(1.0, self.flood_wall_duration_s) # High-substep anti-tunneling precision (20 substeps per frame = 1200 Hz) self.substeps = 20 self.sub_dt = (1.0 / fps) / float(self.substeps) # State tracking self.events: list[SimEvent] = [] self.frames: list[list[FrameRacerTransform]] = [] self.finish_order: list[RacerID] = [] self.finish_frames: dict[RacerID, int] = {} self.per_racer_bounces: dict[RacerID, int] = {r.id: 0 for r in spec.racers} self.last_ranks: dict[RacerID, int] = {r.id: idx + 1 for idx, r in enumerate(spec.racers)} self.orbit_breaks_count = 0 self.stall_breaks_count = 0 # Hazard state (composable layers) track = getattr(spec, "track", None) self.spike_set: set[tuple[int, int]] = set() self.portal_map: dict[tuple[int, int], tuple[int, int]] = {} self.piston_set: set[tuple[int, int]] = set() self.rising_wall: dict | None = None self.hazard_hits = 0 self.portal_uses = 0 self.eliminated: list[RacerID] = [] self.active_spikes: set[tuple[int, int]] = set() if track is not None: try: self.spike_set = {(int(r), int(c)) for (r, c) in list(getattr(track, "spike_cells", []) or [])} self.active_spikes = set(self.spike_set) self.rising_wall = getattr(track, "rising_wall", None) except Exception: pass self._portal_cooldown: dict[RacerID, int] = {r.id: 0 for r in spec.racers} self._spike_cooldown: dict[RacerID, int] = {r.id: 0 for r in spec.racers} _bpm = 128.0 try: _bpm = float(getattr(getattr(spec, "audio", None), "bpm_grid", 128.0)) except Exception: pass self._piston_period_s = (60.0 / max(60.0, _bpm)) * 2.0 def get_flood_wall_y(self, t: float) -> tuple[float, bool]: """Position of the descending rigid flood wall bottom surface at time t.""" if not self.flood_wall_enabled: return -999.0, False return compute_hydraulic_press_y( t=t, y_start=self.flood_wall_y_start, y_finish=self.flood_wall_y_finish, duration_s=self.flood_wall_duration_s, ) def run(self) -> Timeline: """Run entire race using exact theoretical specular reflections and elastic momentum exchange.""" self.active_spikes = set(self.spike_set) fps = self.fps base_speed = self.spec.physics.base_speed gx0, gy0, gx1, gy1 = self.goal_tile rank_changes_count = 0 class RacerState: def __init__(self, rid: RacerID, x: float, y: float, heading_deg: float) -> None: self.id = rid self.x = x self.y = y self.vx = 0.0 self.vy = 0.0 self.heading_deg = heading_deg self.finished = False self.boost_speed = 0.0 self.last_bumper_frame = -20 self.last_flood_bounce_frame = -99 self.consecutive_horiz_bounces = 0 racers = [RacerState(r.id, r.spawn[0], r.spawn[1], r.heading_deg) for r in self.spec.racers] active_racers = list(racers) winner_frame: int | None = None post_win_frames = int(1.5 * fps) # 1.5s victory celebration after winner touches checkered tile for frame in range(self.max_frames): t_curr = frame / float(fps) # 1. Pre-race countdown staging: hold racers at spawn if frame < self.countdown_frames: for r in racers: r.vx = 0.0 r.vy = 0.0 r_spec = next(s for s in self.spec.racers if s.id == r.id) r.x, r.y = r_spec.spawn # Countdown audio/visual trigger ticks: 3, 2, 1 f_08 = int(0.8 * fps) f_17 = int(1.7 * fps) f_26 = int(2.6 * fps) if frame == f_08: self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.COUNTDOWN_TICK, a_id=racers[0].id, meta={"digit": "3"}, ) ) elif frame == f_17: self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.COUNTDOWN_TICK, a_id=racers[0].id, meta={"digit": "2"}, ) ) elif frame == f_26: self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.COUNTDOWN_TICK, a_id=racers[0].id, meta={"digit": "1"}, ) ) f_trans = [] racer_order = [r.id for r in self.spec.racers] for idx, r in enumerate(racers): scell = self.grid.world_to_cell(r.x, r.y) dist = float(self.solver.dist_field[scell.row, scell.col]) if scell else 999.0 f_trans.append( FrameRacerTransform( id=r.id, x=r.x, y=r.y, vx=0.0, vy=0.0, heading_deg=r.heading_deg, distance_to_goal=dist, rank=idx + 1, has_finished=False, is_eliminated=False, ) ) sorted_frame = sorted(f_trans, key=lambda f: racer_order.index(f.id)) self.frames.append(sorted_frame) continue # 2. Launch race at frame == countdown_frames if frame == self.countdown_frames: for r in racers: r_spec = next(s for s in self.spec.racers if s.id == r.id) rad = math.radians(r_spec.heading_deg) r.vx = math.cos(rad) * base_speed r.vy = math.sin(rad) * base_speed self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.RACE_START, a_id=racers[0].id, meta={"digit": "GO"}, ) ) # Compute theoretical constant speed for current frame target_spd = compute_target_speed(self.spec, frame - self.countdown_frames, fps) dt = self.sub_dt hs = self.hs sz = self.sz # Evaluate hydraulic compactor position and stomp triggers cur_yw, is_stomp = self.get_flood_wall_y(t_curr) if self.flood_wall_enabled and is_stomp and frame > self.countdown_frames: cycle_idx = int(t_curr / 4.5) if cycle_idx != self.last_stomp_cycle: self.last_stomp_cycle = cycle_idx self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.PRESS_STOMP, a_id=active_racers[0].id if active_racers else racers[0].id, impulse=800.0, pos=(540.0, cur_yw), meta={"cycle": cycle_idx}, ) ) # 3. Substep mathematical physics loop for _ in range(self.substeps): # Integration and goal check for r in active_racers: if r.finished: continue # Pinball Bumper Collisions (elastic specular launch) for bp_x, bp_y, bp_cell in self.bumper_positions: coll_r = self.hs + self.bumper_radius bdx = r.x - bp_x bdy = r.y - bp_y dist_sq = bdx * bdx + bdy * bdy if dist_sq < coll_r * coll_r: d = math.sqrt(max(1e-8, dist_sq)) bnx = bdx / d bny = bdy / d r.x = bp_x + bnx * (coll_r + 1.0) r.y = bp_y + bny * (coll_r + 1.0) if (frame - r.last_bumper_frame) >= 6: r.last_bumper_frame = frame r.vx = bnx * self.bumper_impulse r.vy = bny * self.bumper_impulse r.boost_speed = self.bumper_impulse r.heading_deg = math.degrees(math.atan2(r.vy, r.vx)) self.bumper_hits += 1 self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.BUMPER_HIT, a_id=r.id, impulse=self.bumper_impulse, normal=(bnx, bny), pos=(bp_x, bp_y), meta={"bumper": f"{bp_cell[0]},{bp_cell[1]}"}, ) ) break prev_x = r.x prev_y = r.y dt = self.sub_dt dx = r.vx * dt dy = r.vy * dt # AABB Checkered Goal Tile Collision if r.x + self.hs >= gx0 and r.x - self.hs <= gx1 and r.y + self.hs >= gy0 and r.y - self.hs <= gy1: r.finished = True r.vx = 0.0 r.vy = 0.0 r.x = self.goal_cx r.y = self.goal_cy self.finish_order.append(r.id) self.finish_frames[r.id] = frame self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.FINISH, a_id=r.id, pos=(self.goal_cx, self.goal_cy), ) ) if winner_frame is None: winner_frame = frame continue # 1. Multi-Bounce Continuous Swept Collision Detection (up to 3 bounces per substep) cell_sz = self.grid.cell_size off_x = self.grid.offset_x off_y = self.grid.offset_y remain_dt = dt for _bounce_iter in range(3): if remain_dt <= 1e-9: break seg_dx = r.vx * remain_dt seg_dy = r.vy * remain_dt seg_dest_x = r.x + seg_dx seg_dest_y = r.y + seg_dy # Check Outer Arena Boundaries & Descending Rigid Flood Wall earliest_t = 1.0 hit_box: WallBox | None = None hit_norm = (0.0, 0.0) hit_is_flood = False eff_min_y = self.arena_min_y cur_yw, _ = self.get_flood_wall_y(t_curr) if self.flood_wall_enabled: eff_min_y = max(self.arena_min_y, cur_yw + self.hs) if seg_dx < -1e-9 and seg_dest_x < self.arena_min_x: t_bound = (self.arena_min_x - r.x) / seg_dx if 0.0 <= t_bound < earliest_t: earliest_t = max(0.0, t_bound) hit_norm = (1.0, 0.0) elif seg_dx > 1e-9 and seg_dest_x > self.arena_max_x: t_bound = (self.arena_max_x - r.x) / seg_dx if 0.0 <= t_bound < earliest_t: earliest_t = max(0.0, t_bound) hit_norm = (-1.0, 0.0) if seg_dy < -1e-9 and seg_dest_y < eff_min_y: t_bound = (eff_min_y - r.y) / seg_dy if 0.0 <= t_bound < earliest_t: earliest_t = max(0.0, t_bound) hit_norm = (0.0, 1.0) if self.flood_wall_enabled and eff_min_y > self.arena_min_y: hit_is_flood = True elif r.y < eff_min_y: earliest_t = 0.0 hit_norm = (0.0, 1.0) if self.flood_wall_enabled and eff_min_y > self.arena_min_y: hit_is_flood = True elif seg_dy > 1e-9 and seg_dest_y > self.arena_max_y: t_bound = (self.arena_max_y - r.y) / seg_dy if 0.0 <= t_bound < earliest_t: earliest_t = max(0.0, t_bound) hit_norm = (0.0, -1.0) # Query Candidate Wall Boxes from Spatial Hash Grid along the current segment c0 = max(0, min(self.grid.cols - 1, int((min(r.x, seg_dest_x) - self.hs - off_x) / cell_sz))) c1 = max(0, min(self.grid.cols - 1, int((max(r.x, seg_dest_x) + self.hs - off_x) / cell_sz))) r0 = max(0, min(self.grid.rows - 1, int((min(r.y, seg_dest_y) - self.hs - off_y) / cell_sz))) r1 = max(0, min(self.grid.rows - 1, int((max(r.y, seg_dest_y) + self.hs - off_y) / cell_sz))) cand_boxes: list[WallBox] = [] seen_boxes: set[int] = set() for c_idx in range(c0, c1 + 1): for r_idx in range(r0, r1 + 1): for box in self.spatial_boxes.get((c_idx, r_idx), []): b_id = id(box) if b_id not in seen_boxes: seen_boxes.add(b_id) cand_boxes.append(box) for box in cand_boxes: bL = box.left - self.hs bR = box.right + self.hs bT = box.top - self.hs bB = box.bottom + self.hs # Initial overlap check if bL < r.x < bR and bT < r.y < bB: ol = r.x - bL orr = bR - r.x ot = r.y - bT ob = bB - r.y m_pen = min(ol, orr, ot, ob) if m_pen == ol: hit_norm = (-1.0, 0.0) elif m_pen == orr: hit_norm = (1.0, 0.0) elif m_pen == ot: hit_norm = (0.0, -1.0) else: hit_norm = (0.0, 1.0) earliest_t = 0.0 hit_box = box break # Continuous slab ray test along X if abs(seg_dx) > 1e-9: if seg_dx > 0: tx1 = (bL - r.x) / seg_dx tx2 = (bR - r.x) / seg_dx nx = -1.0 else: tx1 = (bR - r.x) / seg_dx tx2 = (bL - r.x) / seg_dx nx = 1.0 else: if r.x <= bL or r.x >= bR: continue tx1 = -1e9 tx2 = 1e9 nx = 0.0 # Continuous slab ray test along Y if abs(seg_dy) > 1e-9: if seg_dy > 0: ty1 = (bT - r.y) / seg_dy ty2 = (bB - r.y) / seg_dy ny = -1.0 else: ty1 = (bB - r.y) / seg_dy ty2 = (bT - r.y) / seg_dy ny = 1.0 else: if r.y <= bT or r.y >= bB: continue ty1 = -1e9 ty2 = 1e9 ny = 0.0 t_entry = max(tx1, ty1) t_exit = min(tx2, ty2) if t_entry < t_exit and t_exit > 0.0 and 0.0 <= t_entry < earliest_t: earliest_t = t_entry hit_box = box if tx1 > ty1: hit_norm = (nx, 0.0) else: hit_norm = (0.0, ny) if hit_box is not None or hit_norm != (0.0, 0.0): # Advance racer directly to contact surface r.x += earliest_t * seg_dx r.y += earliest_t * seg_dy remain_dt = (1.0 - earliest_t) * remain_dt # Specular reflection if hit_box is not None: if hit_norm[0] < -0.5: r.x = hit_box.left - self.hs r.vx = -abs(r.vx) elif hit_norm[0] > 0.5: r.x = hit_box.right + self.hs r.vx = abs(r.vx) elif hit_norm[1] < -0.5: r.y = hit_box.top - self.hs r.vy = -abs(r.vy) elif hit_norm[1] > 0.5: r.y = hit_box.bottom + self.hs r.vy = abs(r.vy) if getattr(hit_box, "cell", None) in self.breakable_cells: bk_cell = hit_box.cell self.breakable_cells[bk_cell] -= 1 if self.breakable_cells[bk_cell] <= 0: self.blocks_shattered += 1 del self.breakable_cells[bk_cell] self.grid.grid[bk_cell[0], bk_cell[1]] = CellType.FREE if hit_box in self.wall_boxes: self.wall_boxes.remove(hit_box) for clist in self.spatial_boxes.values(): if hit_box in clist: clist.remove(hit_box) self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.BLOCK_SHATTER, a_id=r.id, impulse=target_spd, pos=((hit_box.left + hit_box.right) / 2.0, (hit_box.top + hit_box.bottom) / 2.0), meta={"cell": f"{bk_cell[0]},{bk_cell[1]}"}, ) ) else: if hit_norm[0] > 0.5: r.x = self.arena_min_x r.vx = abs(r.vx) elif hit_norm[0] < -0.5: r.x = self.arena_max_x r.vx = -abs(r.vx) elif hit_norm[1] > 0.5: if hit_is_flood: r.y = eff_min_y + 0.5 r.vy = abs(r.vy) + 1.2 * self.flood_wall_vy gx, _ = self.solver.get_gradient_toward_goal(r.x, r.y) wedge_dir = 1.0 if gx >= 0 else -1.0 r.vx += wedge_dir * 0.40 * target_spd cur_spd = math.hypot(r.vx, r.vy) if cur_spd > 1e-6: r.vx = (r.vx / cur_spd) * target_spd r.vy = (r.vy / cur_spd) * target_spd else: r.y = self.arena_min_y r.vy = abs(r.vy) elif hit_norm[1] < -0.5: r.y = self.arena_max_y r.vy = -abs(r.vy) # Anti-Stall & Anti-Vertical Bouncing Guard is_horiz_hit = abs(hit_norm[1]) > 0.5 if is_horiz_hit: r.consecutive_horiz_bounces += 1 # Enforce minimum lateral slice angle so racer never bounces in pure vertical 1D lock min_lat = target_spd * 0.38 if abs(r.vx) < min_lat: gx, gy = self.solver.get_gradient_toward_goal(r.x, r.y) lat_dir = 1.0 if gx > 0.05 else (-1.0 if gx < -0.05 else (1.0 if r.x < 540.0 else -1.0)) r.vx = lat_dir * min_lat rem_sq = max(100.0, target_spd * target_spd - r.vx * r.vx) r.vy = math.copysign(math.sqrt(rem_sq), r.vy) # Tangential surface offset deflection (natural flare off box edge) if hit_box is not None and hit_box.w > 10.0: box_mid_x = (hit_box.left + hit_box.right) / 2.0 edge_frac = (r.x - box_mid_x) / (hit_box.w / 2.0) edge_frac = max(-1.0, min(1.0, edge_frac)) r.vx += edge_frac * 0.12 * target_spd c_spd = math.hypot(r.vx, r.vy) if c_spd > 1e-4: r.vx = (r.vx / c_spd) * target_spd r.vy = (r.vy / c_spd) * target_spd # Anti-vertical deflection slice if bouncing between horizontal surfaces 3+ times if r.consecutive_horiz_bounces >= 3: r.consecutive_horiz_bounces = 0 gx, gy = self.solver.get_gradient_toward_goal(r.x, r.y) grad_angle = math.atan2(gy, gx) r.vx = math.cos(grad_angle) * target_spd r.vy = math.sin(grad_angle) * target_spd else: r.consecutive_horiz_bounces = 0 meta_dict: dict[str, Any] = {} should_emit_bounce = True if hit_is_flood: meta_dict["wall_type"] = "flood_barrier" # Check if racer is pinched/crushed against a static block or arena floor below is_crushed = False c_idx = max(0, min(self.grid.cols - 1, int((r.x - off_x) / cell_sz))) r_idx = max(0, min(self.grid.rows - 1, int((r.y + self.hs - off_y) / cell_sz))) for box in self.spatial_boxes.get((c_idx, r_idx), []): if box.left - self.hs <= r.x <= box.right + self.hs: if box.top > cur_yw: vert_gap = box.top - cur_yw if vert_gap < self.sz * 0.72: is_crushed = True break else: # Nudge sideways off the ledge toward the open channel box_mid_x = (box.left + box.right) / 2.0 nudge_dir = 1.0 if r.x >= box_mid_x else -1.0 r.x += nudge_dir * (self.hs * 0.8) r.vx = nudge_dir * abs(r.vx or target_spd) cur_spd = math.hypot(r.vx, r.vy) if cur_spd > 1e-6: r.vx = (r.vx / cur_spd) * target_spd r.vy = (r.vy / cur_spd) * target_spd break if (self.arena_max_y - cur_yw) < self.sz * 0.72: is_crushed = True if is_crushed: r.finished = True r.vx = 0.0 r.vy = 0.0 self.eliminated.append(r.id) self.hazard_hits += 1 self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.HAZARD_WALL, a_id=r.id, pos=(r.x, cur_yw + self.hs * 0.4), meta={"eliminated": "crushed", "wall_type": "flood_barrier"}, ) ) remain_dt = 0.0 break # Debounce compactor contact sounds: at most once every 12 frames (200ms) if frame - getattr(r, "last_flood_bounce_frame", -99) < 12: should_emit_bounce = False else: r.last_flood_bounce_frame = frame if should_emit_bounce: self.per_racer_bounces[r.id] += 1 self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.WALL_BOUNCE, a_id=r.id, impulse=target_spd, normal=hit_norm, pos=(r.x, r.y), meta=meta_dict, ) ) else: r.x = seg_dest_x r.y = seg_dest_y remain_dt = 0.0 break # 4. Multi-pass SAT De-penetration pass (resolves any residual corner overlap) for _sat_iter in range(2): c0 = max(0, min(self.grid.cols - 1, int((r.x - self.hs - off_x) / cell_sz))) c1 = max(0, min(self.grid.cols - 1, int((r.x + self.hs - off_x) / cell_sz))) r0 = max(0, min(self.grid.rows - 1, int((r.y - self.hs - off_y) / cell_sz))) r1 = max(0, min(self.grid.rows - 1, int((r.y + self.hs - off_y) / cell_sz))) local_boxes: list[WallBox] = [] seen_local: set[int] = set() for c_idx in range(c0, c1 + 1): for r_idx in range(r0, r1 + 1): for box in self.spatial_boxes.get((c_idx, r_idx), []): b_id = id(box) if b_id not in seen_local: seen_local.add(b_id) local_boxes.append(box) had_penetration = False for box in local_boxes: bL = box.left - self.hs bR = box.right + self.hs bT = box.top - self.hs bB = box.bottom + self.hs if bL < r.x < bR and bT < r.y < bB: ol = r.x - bL orr = bR - r.x ot = r.y - bT ob = bB - r.y m_pen = min(ol, orr, ot, ob) if m_pen == ol: r.x = bL r.vx = -abs(r.vx) elif m_pen == orr: r.x = bR r.vx = abs(r.vx) elif m_pen == ot: r.y = bT r.vy = -abs(r.vy) else: r.y = bB r.vy = abs(r.vy) had_penetration = True if not had_penetration: break # 5. Grid cell 4-corner containment fail-safe (smooth push-out instead of teleporting) corners = [ (r.x - self.hs, r.y - self.hs), (r.x + self.hs, r.y - self.hs), (r.x - self.hs, r.y + self.hs), (r.x + self.hs, r.y + self.hs), ] for cx, cy in corners: ccell = self.grid.world_to_cell(cx, cy) if ccell and self.grid.grid[ccell.row, ccell.col] == CellType.WALL: w_x0, w_y0, w_x1, w_y1 = self.grid.cell_bounds(ccell.col, ccell.row) d_left = abs(cx - w_x0) d_right = abs(w_x1 - cx) d_top = abs(cy - w_y0) d_bot = abs(w_y1 - cy) min_d = min(d_left, d_right, d_top, d_bot) if min_d == d_left: r.x = w_x0 - self.hs r.vx = -abs(r.vx) elif min_d == d_right: r.x = w_x1 + self.hs r.vx = abs(r.vx) elif min_d == d_top: r.y = w_y0 - self.hs r.vy = -abs(r.vy) else: r.y = w_y1 + self.hs r.vy = abs(r.vy) # 6. Strict Arena Outer & Flood Wall Clamp r.x = max(self.arena_min_x, min(self.arena_max_x, r.x)) r.y = max(eff_min_y, min(self.arena_max_y, r.y)) if self.flood_wall_enabled and not (r.finished or r.id in self.eliminated): if cur_yw >= r.y - self.hs * 0.25: r.finished = True r.vx = 0.0 r.vy = 0.0 self.eliminated.append(r.id) self.hazard_hits += 1 self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.HAZARD_WALL, a_id=r.id, pos=(r.x, cur_yw + self.hs * 0.4), meta={"eliminated": "crushed", "wall_type": "flood_barrier"}, ) ) continue elif r.y <= eff_min_y and r.vy < 0: r.vy = abs(r.vy) # One-Shot Lethal Corner Spike Check (Instant Elimination + Spike Disappearance) if self.active_spikes: hit_spike = None for sp_row, sp_col in list(self.active_spikes): sx0, sy0, sx1, sy1 = self.grid.cell_bounds(sp_col, sp_row) if ( r.x + self.hs > sx0 + 4.0 and r.x - self.hs < sx1 - 4.0 and r.y + self.hs > sy0 + 4.0 and r.y - self.hs < sy1 - 4.0 ): hit_spike = (sp_row, sp_col) break if hit_spike is not None: r.finished = True r.vx = 0.0 r.vy = 0.0 self.eliminated.append(r.id) self.active_spikes.discard(hit_spike) self.hazard_hits += 1 self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.HAZARD_SPIKE, a_id=r.id, impulse=target_spd, pos=(r.x, r.y), meta={"eliminated": "true", "cell": f"{hit_spike[0]},{hit_spike[1]}"}, ) ) continue # 7. Maintain speed invariant with smooth boost decay if getattr(r, "boost_speed", 0.0) > target_spd: r.boost_speed = max(target_spd, r.boost_speed * math.pow(0.985, self.sub_dt * 60.0)) eff_spd = max(target_spd, getattr(r, "boost_speed", target_spd)) cur_spd = math.hypot(r.vx, r.vy) if cur_spd > 1e-4: r.vx = (r.vx / cur_spd) * eff_spd r.vy = (r.vy / cur_spd) * eff_spd r.heading_deg = math.degrees(math.atan2(r.vy, r.vx)) # Exact Center-of-Mass Elastic Square-on-Square Collisions for i in range(len(active_racers)): for j in range(i + 1, len(active_racers)): r1 = active_racers[i] r2 = active_racers[j] if r1.finished or r2.finished: continue dx = r2.x - r1.x dy = r2.y - r1.y if abs(dx) < sz and abs(dy) < sz: ox = sz - abs(dx) oy = sz - abs(dy) if ox < oy: nx = 1.0 if dx > 0 else -1.0 ny = 0.0 overlap = ox else: nx = 0.0 ny = 1.0 if dy > 0 else -1.0 overlap = oy v_rel_n = (r1.vx - r2.vx) * nx + (r1.vy - r2.vy) * ny if v_rel_n > 0: # Classical elastic impulse exchange in center-of-mass frame cm_x = 0.5 * (r1.vx + r2.vx) cm_y = 0.5 * (r1.vy + r2.vy) u_x = 0.5 * (r1.vx - r2.vx) u_y = 0.5 * (r1.vy - r2.vy) dot = u_x * nx + u_y * ny u_prime_x = u_x - 2.0 * dot * nx u_prime_y = u_y - 2.0 * dot * ny v1x = cm_x + u_prime_x v1y = cm_y + u_prime_y v2x = cm_x - u_prime_x v2y = cm_y - u_prime_y l1 = math.hypot(v1x, v1y) l2 = math.hypot(v2x, v2y) r1_eff = max(target_spd, getattr(r1, "boost_speed", target_spd)) r2_eff = max(target_spd, getattr(r2, "boost_speed", target_spd)) r1.vx = (v1x / l1) * r1_eff if l1 > 1e-4 else r1.vx r1.vy = (v1y / l1) * r1_eff if l1 > 1e-4 else r1.vy r2.vx = (v2x / l2) * r2_eff if l2 > 1e-4 else r2.vx r2.vy = (v2y / l2) * r2_eff if l2 > 1e-4 else r2.vy r1.heading_deg = math.degrees(math.atan2(r1.vy, r1.vx)) r2.heading_deg = math.degrees(math.atan2(r2.vy, r2.vx)) self.per_racer_bounces[r1.id] += 1 self.per_racer_bounces[r2.id] += 1 self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.RACER_BOUNCE, a_id=r1.id, b_id=r2.id, impulse=v_rel_n, pos=(0.5 * (r1.x + r2.x), 0.5 * (r1.y + r2.y)), ) ) # Non-penetration positional separation along contact normal r1.x -= nx * overlap * 0.5 r1.y -= ny * overlap * 0.5 r2.x += nx * overlap * 0.5 r2.y += ny * overlap * 0.5 # Ensure racer separation never pushes either racer into a wall or outer boundary for r_sep in (r1, r2): r_sep.x = max(self.arena_min_x, min(self.arena_max_x, r_sep.x)) r_sep.y = max(self.arena_min_y, min(self.arena_max_y, r_sep.y)) c_sc0 = max(0, min(self.grid.cols - 1, int((r_sep.x - self.hs - off_x) / cell_sz))) c_sc1 = max(0, min(self.grid.cols - 1, int((r_sep.x + self.hs - off_x) / cell_sz))) r_sr0 = max(0, min(self.grid.rows - 1, int((r_sep.y - self.hs - off_y) / cell_sz))) r_sr1 = max(0, min(self.grid.rows - 1, int((r_sep.y + self.hs - off_y) / cell_sz))) for c_idx in range(c_sc0, c_sc1 + 1): for r_idx in range(r_sr0, r_sr1 + 1): for box in self.spatial_boxes.get((c_idx, r_idx), []): bL = box.left - self.hs bR = box.right + self.hs bT = box.top - self.hs bB = box.bottom + self.hs if bL < r_sep.x < bR and bT < r_sep.y < bB: ol = r_sep.x - bL orr = bR - r_sep.x ot = r_sep.y - bT ob = bB - r_sep.y m_pen = min(ol, orr, ot, ob) if m_pen == ol: r_sep.x = bL elif m_pen == orr: r_sep.x = bR elif m_pen == ot: r_sep.y = bT else: r_sep.y = bB # 3b. Hazard runtime hooks (rising wall) if self.rising_wall is not None and not self.flood_wall_enabled: try: from squarecore.track.hazards import rising_wall_y as _rwy wy = _rwy(self.grid, t_curr, self.rising_wall) direction = str(self.rising_wall.get("direction", "top_down")) for r in active_racers: if r.finished or r.id in self.eliminated: continue hit = (r.y <= wy) if direction == "top_down" else (r.y >= wy) if hit and t_curr > self.countdown_s + 12.0: r.finished = True r.vx, r.vy = 0.0, 0.0 self.eliminated.append(r.id) self.hazard_hits += 1 self.events.append(SimEvent(frame=frame, t=t_curr, kind=EventKind.HAZARD_WALL, a_id=r.id, pos=(r.x, r.y))) except Exception: pass # 4. Rank tracking and frame transform recording distances: dict[RacerID, float] = {} for r in racers: if r.id in self.eliminated: dist = 99999.0 elif r.finished: dist = -1.0 else: cell = self.grid.world_to_cell(r.x, r.y) dist = float(self.solver.dist_field[cell.row, cell.col]) if cell else 999.0 distances[r.id] = dist sorted_by_dist = sorted( racers, key=lambda r: ( r.id in self.eliminated, not (r.finished and r.id not in self.eliminated), distances[r.id], ), ) for rank_idx, r in enumerate(sorted_by_dist): cur_rank = rank_idx + 1 if cur_rank != self.last_ranks[r.id]: if not r.finished and r.id not in self.eliminated: rank_changes_count += 1 self.events.append( SimEvent( frame=frame, t=t_curr, kind=EventKind.RANK_CHANGE, a_id=r.id, meta={"old_rank": self.last_ranks[r.id], "new_rank": cur_rank}, ) ) self.last_ranks[r.id] = cur_rank f_trans = [] racer_order = [r.id for r in self.spec.racers] for r in sorted(racers, key=lambda r: racer_order.index(r.id)): f_trans.append( FrameRacerTransform( id=r.id, x=r.x, y=r.y, vx=r.vx, vy=r.vy, heading_deg=r.heading_deg, distance_to_goal=distances[r.id], rank=self.last_ranks[r.id], has_finished=r.finished and (r.id not in self.eliminated), is_eliminated=(r.id in self.eliminated), ) ) self.frames.append(f_trans) # Check post-win cutoff (1.5s after winner crosses line) if winner_frame is not None and (frame - winner_frame) >= post_win_frames: break tot_frames = len(self.frames) tot_dur = tot_frames / float(fps) winner = self.finish_order[0] if self.finish_order else None tot_bounces = sum(self.per_racer_bounces.values()) outcome = SimOutcome( winner=winner, finish_order=self.finish_order, finish_frames=self.finish_frames, rank_changes=rank_changes_count, total_bounces=tot_bounces, per_racer_bounces=self.per_racer_bounces, orbit_breaks=self.orbit_breaks_count, stall_breaks=self.stall_breaks_count, total_frames=tot_frames, total_duration_s=tot_dur, hazard_hits=self.hazard_hits, portal_uses=self.portal_uses, bumper_hits=self.bumper_hits, blocks_shattered=self.blocks_shattered, eliminated=list(self.eliminated), ) return Timeline( seed=self.spec.seed, fps=fps, total_frames=tot_frames, frames=self.frames, events=self.events, outcome=outcome, )