multiplayer_js.py

import asyncio
import json
import time


# =========================
# CONFIG
# =========================
HOST = "0.0.0.0"
PORT = 4445

TICK_RATE = 20
PING_INTERVAL = 1.0
PLAYER_TIMEOUT = 5.0
MAX_PACKET_SIZE = 1200


# =========================
# PLAYER CLASS
# =========================
class Player:
    def __init__(self, addr):
        self.addr = addr
        self.last_seen = time.time()

        # position
        self.x = 0.0
        self.y = 0.0
        self.z = 0.0

        self.px = 0.0
        self.py = 0.0
        self.pz = 0.0

        # stats
        self.health = 100
        self.armor = 0

        # weapons
        self.weapon_id = 0
        self.primary_ammo = 30
        self.secondary_ammo = 90

        # movement
        self.walking = False
        self.running = False

        # network
        self.ping = 0.0


    # -------------------------
    # APPLY MOVEMENT
    # -------------------------
    def apply_movement(self, msg):
        self.px, self.py, self.pz = self.x, self.y, self.z

        self.x += float(msg.get("dx", 0))
        self.y += float(msg.get("dy", 0))
        self.z += float(msg.get("dz", 0))

        self.walking = bool(msg.get("walking", False))
        self.running = bool(msg.get("running", False))


    # -------------------------
    # APPLY COMBAT
    # -------------------------
    def apply_combat(self, msg):
        self.weapon_id = int(msg.get("weapon_id", self.weapon_id))
        self.primary_ammo = int(msg.get("primary_ammo", self.primary_ammo))
        self.secondary_ammo = int(msg.get("secondary_ammo", self.secondary_ammo))


    # -------------------------
    # APPLY DAMAGE
    # -------------------------
    def apply_damage(self, msg):
        dmg = int(msg.get("amount", 0))

        absorbed = min(self.armor, dmg)
        self.armor -= absorbed
        dmg -= absorbed

        self.health = max(0, self.health - dmg)


    # -------------------------
    # APPLY PING
    # -------------------------
    def apply_ping(self, msg):
        client_time = float(msg.get("t", time.time()))
        self.ping = time.time() - client_time


    # -------------------------
    # UPDATE ACTIVITY
    # -------------------------
    def touch(self):
        self.last_seen = time.time()


    # -------------------------
    # SERIALIZE (send to clients)
    # -------------------------
    def serialize(self):
        return {
            "x": self.x,
            "y": self.y,
            "z": self.z,

            "px": self.px,
            "py": self.py,
            "pz": self.pz,

            "health": self.health,
            "armor": self.armor,

            "weapon_id": self.weapon_id,
            "primary_ammo": self.primary_ammo,
            "secondary_ammo": self.secondary_ammo,

            "walking": self.walking,
            "running": self.running,

            "ping": round(self.ping, 3)
        }


# =========================
# GAME SERVER CLASS
# =========================
class GameServer(asyncio.DatagramProtocol):
    def __init__(self):
        self.players = {}  # addr -> Player


    # -------------------------
    # GET OR CREATE PLAYER
    # -------------------------
    def get_player(self, addr):
        if addr not in self.players:
            self.players[addr] = Player(addr)
        return self.players[addr]


    # -------------------------
    # PARSE PACKET SAFELY
    # -------------------------
    def parse(self, data):
        if len(data) > MAX_PACKET_SIZE:
            return None
        try:
            return json.loads(data)
        except:
            return None


    # -------------------------
    # HANDLE INCOMING PACKETS
    # -------------------------
    def datagram_received(self, data, addr):
        msg = self.parse(data)
        if not msg:
            return

        player = self.get_player(addr)
        player.touch()

        mtype = msg.get("type")

        if mtype == "update":
            player.apply_movement(msg)

        elif mtype == "combat":
            player.apply_combat(msg)

        elif mtype == "damage":
            player.apply_damage(msg)

        elif mtype == "ping":
            player.apply_ping(msg)


    # -------------------------
    # REMOVE DISCONNECTED PLAYERS
    # -------------------------
    def cleanup_players(self):
        now = time.time()
        dead = []

        for addr, player in self.players.items():
            if now - player.last_seen > PLAYER_TIMEOUT:
                dead.append(addr)

        for addr in dead:
            del self.players[addr]


    # -------------------------
    # BUILD WORLD STATE
    # -------------------------
    def build_world(self):
        return {
            "type": "world",
            "players": {
                str(addr): p.serialize()
                for addr, p in self.players.items()
            }
        }


# =========================
# BROADCAST LOOP
# =========================
async def broadcast_loop(server):
    interval = 1.0 / TICK_RATE

    while True:
        start = time.time()

        server.cleanup_players()

        world = server.build_world()
        packet = json.dumps(world).encode()

        dead = []

        for addr in server.players:
            try:
                server.transport.sendto(packet, addr)
            except:
                dead.append(addr)

        for d in dead:
            server.players.pop(d, None)

        elapsed = time.time() - start
        await asyncio.sleep(max(0, interval - elapsed))


# =========================
# PING / TIME SYSTEM
# =========================
async def ping_loop(server):
    while True:
        # could extend for latency smoothing logic
        await asyncio.sleep(PING_INTERVAL)


# =========================
# START SERVER
# =========================
async def main():
    loop = asyncio.get_running_loop()

    server = GameServer()

    transport, _ = await loop.create_datagram_endpoint(
        lambda: server,
        local_addr=(HOST, PORT)
    )

    server.transport = transport

    print(f"Class-based game server running on {HOST}:{PORT}")

    await asyncio.gather(
        broadcast_loop(server),
        ping_loop(server)
    )


asyncio.run(main())