""" AI PLATFORMER + CHATBOT (RICH GRAPHICS + FIXED PHYSICS) AABB Collision, Detailed Rendering, Stateful Engine """ import os, json, random, threading, logging, time from collections import deque from dataclasses import dataclass from typing import Dict, List, Optional import numpy as np import torch import torch.nn as nn import torch.optim as optim from flask import Flask, jsonify, request, render_template_string logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger(__name__) @dataclass class Cfg: W: int = 80; H: int = 20; GROUND: int = 17; CHUNK: int = 30 SAFE: int = 15; VIEW: int = 40 GRAV: float = 0.35; JUMP: float = -6.5; SPEED: float = 0.35 STATE: int = 40 * 40; ACTS: int = 4; MEM: int = 10000 BATCH: int = 64; GAMMA: float = 0.99; LR: float = 5e-4 EPS_DEC: float = 0.995; PORT: int = 7860 MODEL: str = "dqn_model.pth"; CHAT: str = "chat_data.json" C = Cfg() # ============================================================================ # GAME ENGINE: AABB PHYSICS + RICH WORLD # ============================================================================ class Engine: def __init__(self, seed=None): self.seed = seed or random.randint(0, 999999) self.reset() def reset(self): self.px, self.py = 5.0, float(C.GROUND) self.vx, self.vy = 0.0, 0.0 self.grounded = True self.alive = True self.score = 0 self.coins = 0 self.step_n = 0 self.chunks: Dict[int, dict] = {} self.obs: List[dict] = [] self.enemies: List[dict] = [] self.coin_list: List[dict] = [] self._load_chunks() return self.get_state() def _gen_chunk(self, cid: int) -> dict: rng = random.Random((cid * 1337 + self.seed) % 999999) bx = cid * C.CHUNK obs, ens, cns = [], [], [] diff = max(1.0, abs(cid) * 0.1) safe = bx < C.SAFE if not safe: for _ in range(rng.randint(3, 6) + int(diff)): x = bx + rng.randint(5, 25) h = rng.randint(1, 3 + int(diff * 0.5)) w = rng.randint(1, 3) obs.append({'x': x, 'y': C.GROUND - h, 'w': w, 'h': h, 'pit': False}) for _ in range(rng.randint(1, 2)): x = bx + rng.randint(10, 20) obs.append({'x': x, 'y': C.GROUND + 1, 'w': rng.randint(2, 4), 'h': 1, 'pit': True}) for _ in range(rng.randint(1, 2)): x = bx + rng.randint(10, 20) ens.append({ 'x': x, 'y': C.GROUND - 1, 'type': rng.choice(['walker', 'jumper']), 'dir': rng.choice([-1, 1]), 'spd': 0.3 + rng.random() * 0.3, 'rng': rng.randint(3, 8), 'ox': x }) for _ in range(rng.randint(5, 10) + int(diff)): cns.append({ 'x': bx + rng.randint(2, 28), 'y': rng.randint(5, C.GROUND - 2), 'collected': False }) return {'obs': obs, 'ens': ens, 'cns': cns} def _load_chunks(self): cc = int(self.px // C.CHUNK) for i in range(cc - 1, cc + 3): if i not in self.chunks: self.chunks[i] = self._gen_chunk(i) vl, vr = self.px - C.W / 2, self.px + C.W / 2 self.obs, self.enemies, self.coin_list = [], [], [] for i in range(cc - 1, cc + 3): ch = self.chunks.get(i, {}) self.obs.extend([o for o in ch.get('obs', []) if vl <= o['x'] <= vr]) self.enemies.extend([e for e in ch.get('ens', []) if vl <= e['x'] <= vr]) self.coin_list.extend([c for c in ch.get('cns', []) if not c['collected'] and vl <= c['x'] <= vr]) def _aabb(self, ax, ay, aw, ah, bx, by, bw, bh): return ax < bx + bw and ax + aw > bx and ay < by + bh and ay + ah > by def get_state(self): s = np.zeros((C.VIEW, C.VIEW), dtype=np.float32) h = C.VIEW // 2 px, py = int(round(self.px)), int(round(self.py)) s[h, h] = 1.0 for o in self.obs: dx, dy = int(round(o['x'])) - px, int(round(o['y'])) - py v = -1.0 if o.get('pit') else 0.8 for ww in range(o.get('w', 1)): for hh in range(o.get('h', 1)): sx, sy = h + dx + ww, h + dy + hh if 0 <= sx < C.VIEW and 0 <= sy < C.VIEW: s[sy, sx] = v for e in self.enemies: dx, dy = int(round(e['x'])) - px, int(round(e['y'])) - py if 0 <= h + dx < C.VIEW and 0 <= h + dy < C.VIEW: s[h + dy, h + dx] = 0.7 for c in self.coin_list: dx, dy = int(round(c['x'])) - px, int(round(c['y'])) - py if 0 <= h + dx < C.VIEW and 0 <= h + dy < C.VIEW: s[h + dy, h + dx] = 0.3 return s.flatten() def step(self, action: int): sound = None PW, PH = 0.6, 0.9 # Player hitbox size # Input self.vx = 0.0 if action == 1: self.vx = -C.SPEED elif action == 2: self.vx = C.SPEED if action == 3 and self.grounded: self.vy = C.JUMP self.grounded = False sound = 'jump' # === X AXIS MOVEMENT + COLLISION === self.px += self.vx for o in self.obs: if o.get('pit'): continue if self._aabb(self.px, self.py, PW, PH, o['x'], o['y'], o['w'], o['h']): if self.vx > 0: self.px = o['x'] - PW elif self.vx < 0: self.px = o['x'] + o['w'] self.vx = 0 # === Y AXIS MOVEMENT + COLLISION === self.vy += C.GRAV self.py += self.vy self.grounded = False # Ground collision if self.py >= C.GROUND: self.py = C.GROUND self.vy = 0.0 self.grounded = True # Platform collision (Y) for o in self.obs: if o.get('pit'): continue if self._aabb(self.px, self.py, PW, PH, o['x'], o['y'], o['w'], o['h']): if self.vy > 0: # Falling down onto platform self.py = o['y'] - PH self.vy = 0.0 self.grounded = True elif self.vy < 0: # Jumping up into platform self.py = o['y'] + o['h'] self.vy = 0.0 # Death: fell off world if self.py > C.H + 2: self.alive = False return self.get_state(), -50.0, True, 'die' # Pit death for o in self.obs: if o.get('pit') and o['x'] <= self.px + PW / 2 <= o['x'] + o['w'] and self.py >= C.GROUND: self.alive = False return self.get_state(), -50.0, True, 'die' # Enemy collision for e in self.enemies: if self._aabb(self.px, self.py, PW, PH, e['x'] - 0.3, e['y'] - 0.3, 0.6, 0.6): self.alive = False return self.get_state(), -50.0, True, 'die' # Coins got = 0 for c in self.coin_list: if not c['collected'] and self._aabb(self.px, self.py, PW, PH, c['x'] - 0.3, c['y'] - 0.3, 0.6, 0.6): c['collected'] = True got += 1 if got: self.coins += got self.score += got * 10 sound = 'coin' # Update enemies t = time.time() for e in self.enemies: if e['type'] == 'walker': e['x'] += e['spd'] * e['dir'] if abs(e['x'] - e['ox']) > e['rng']: e['dir'] *= -1 else: e['y'] = (C.GROUND - 1) + np.sin(t * e['spd'] * 3) * 0.5 self.score += 1 self.step_n += 1 self._load_chunks() done = self.step_n > 3000 reward = 1.0 + got * 5.0 return self.get_state(), reward, done, sound def world_data(self): return { 'player': [round(self.px, 2), round(self.py, 2)], 'obstacles': self.obs, 'entities': self.enemies, 'coins': [c for c in self.coin_list if not c['collected']], 'ground': C.GROUND, 'score': self.score, 'coins_collected': self.coins, 'alive': self.alive } # ============================================================================ # DQN AGENT # ============================================================================ class Net(nn.Module): def __init__(self): super().__init__() self.net = nn.Sequential( nn.Linear(C.STATE, 256), nn.ReLU(), nn.Linear(256, 256), nn.ReLU(), nn.Linear(256, 128), nn.ReLU(), nn.Linear(128, C.ACTS) ) def forward(self, x): return self.net(x) class Agent: def __init__(self): self.dev = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.model = Net().to(self.dev) self.target = Net().to(self.dev) self.target.load_state_dict(self.model.state_dict()) self.opt = optim.Adam(self.model.parameters(), lr=C.LR) self.crit = nn.MSELoss() self.mem = deque(maxlen=C.MEM) self.eps = 1.0; self.steps = 0; self.best = 0; self.training = False if os.path.exists(C.MODEL): try: self.model.load_state_dict(torch.load(C.MODEL, map_location=self.dev)) self.target.load_state_dict(self.model.state_dict()) logger.info("✅ Model loaded") except Exception as e: logger.warning(f"⚠️ Load failed: {e}") def act(self, s): if random.random() <= self.eps: return random.randrange(C.ACTS) with torch.no_grad(): return torch.argmax(self.model(torch.FloatTensor(s).unsqueeze(0).to(self.dev))).item() def remember(self, s, a, r, ns, d): self.mem.append((s, a, r, ns, d)) def replay(self): if len(self.mem) < C.BATCH: return b = random.sample(self.mem, C.BATCH) st = torch.FloatTensor([x[0] for x in b]).to(self.dev) ac = torch.LongTensor([x[1] for x in b]).to(self.dev) rw = torch.FloatTensor([x[2] for x in b]).to(self.dev) ns = torch.FloatTensor([x[3] for x in b]).to(self.dev) dn = torch.FloatTensor([x[4] for x in b]).to(self.dev) q = self.model(st).gather(1, ac.unsqueeze(1)).squeeze() nq = self.target(ns).max(1)[0].detach() tgt = rw + C.GAMMA * nq * (1 - dn) loss = self.crit(q, tgt) self.opt.zero_grad(); loss.backward(); self.opt.step() if self.eps > 0.01: self.eps *= C.EPS_DEC self.steps += 1 if self.steps % 100 == 0: self.target.load_state_dict(self.model.state_dict()) def train_ep(self): env = Engine(); s = env.reset(); tr = 0.0; d = False; n = 0 while not d and n < 500: a = self.act(s); ns, r, d, _ = env.step(a) self.remember(s, a, r, ns, d); self.replay() s = ns; tr += r; n += 1 if tr > self.best: self.best = tr; self.save() return tr def save(self): torch.save(self.model.state_dict(), C.MODEL) # ============================================================================ # CHAT MEMORY # ============================================================================ class ChatMem: def __init__(self): self.data = {} if os.path.exists(C.CHAT): try: with open(C.CHAT, 'r', encoding='utf-8') as f: self.data = json.load(f) except: pass def save(self): with open(C.CHAT, 'w', encoding='utf-8') as f: json.dump(self.data, f, ensure_ascii=False, indent=2) def add(self, q, a): self.data[q.lower()] = a; self.save(); return f"✅ {q} → {a}" def find(self, q): q = q.lower() if q in self.data: return self.data[q] words = q.split(); best, bs = None, 0 for k, v in self.data.items(): sc = sum(1 for w in words if w in k) if sc > bs: bs, best = sc, v return best if bs >= len(words) * 0.4 else None # ============================================================================ # GLOBAL STATE # ============================================================================ agent = Agent() chat = ChatMem() seed = random.randint(0, 999999) ai_env = Engine(seed) pl_env = Engine(seed) is_training = False # ============================================================================ # RICH GRAPHICS HTML # ============================================================================ HTML = """
🤖 Нейросеть слева 🎮 Ты справа (⬅️ ➡️ ⬆️)
DQN (256→256→128 нейронов)