Spaces:
Running
Running
Antigravity
fix(audio): eliminate compactor squeeze chatter, add pinch elimination & debounced sound
15b28e1 | """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, | |
| ) | |