giantsd/server.py

92 lines
3.3 KiB
Python

from giants.map import Map
from giants.player import Player
from dpnet.netserver import Netserver
from dpnet.session import Session
import socket
import importlib
import os
from giants import Teams, GameTypes
from utils.logger import setup_logger
import traceback
logger = setup_logger(__name__)
class Server:
def __init__(self, **kwargs):
self.listen_ip = kwargs.get("ip", "0.0.0.0")
self.listen_port = kwargs.get("port", 19711)
fake_session = Session(socket.socket(socket.AF_INET, socket.SOCK_DGRAM))
fake_session.ip = "127.0.0.1"
fake_session.port = 3333
self.players = [Player("Amazed4", fake_session)]
self.currentmap = kwargs.get("map", Map("Three Way Island - Canyons.gck"))
self.maxplayers = kwargs.get("maxplayers", 20)
self.name = kwargs.get("name", "Default Server Name")
self.accept_new_players = True
self.register_with_ms = kwargs.get("register", False)
self.teams = kwargs.get("teams", Teams.MvM)
self.game_type = kwargs.get("gametype", GameTypes.TeamDeathmatchWithFullBase)
self.version = 1.497
self.points_per_kill = 1
self.points_per_capture = 5
self.detente_time = 0 # minutes
self._plugins = []
def add_player(self, player):
self.players.append(player)
self._broadcast_event("on_player_join", player)
def add_plugin(self, plugin):
self._plugins.append(plugin)
def change_map(self, mappath):
try:
self.currentmap = Map(mappath)
self._broadcast_event("on_map_change", self.currentmap)
except Exception:
logger.error("Could not change map")
def _broadcast_event(self, event, *args):
logger.debug("Broadcasting event "+event)
for plugin in self._plugins:
if hasattr(plugin, event):
func = getattr(plugin, event)
if callable(func):
try:
func(args)
except Exception:
logger.error("Could not call plugin function: "+plugin.__name__+"."+event)
traceback.print_exc()
def load_plugins(self):
plugins = os.listdir("plugins")
for plugin in plugins:
if os.path.isdir("plugins/"+plugin) or plugin == "__init__.py":
continue
if not plugin.endswith(".py"):
continue
pluginname = plugin.split(".py")[0]
try:
module = importlib.import_module("plugins."+pluginname)
if hasattr(module, "setup"):
setup = getattr(module, "setup")
if callable(setup):
logger.info("Loading plugin "+module.__name__)
module.setup(self)
except Exception:
logger.warn("Could not load plugin "+plugin)
traceback.print_exc()
def broadcast_message(self, text):
for player in self.players:
player.session.send_gamedata(b'\x35\x80\x81'+text.encode("ascii")+b'\x00\x00')
def run(self):
self.load_plugins()
return Netserver(self).run()
if __name__ == '__main__':
server = Server(name="giantsd", maxplayers=10, register=False)
server.run() # blocking