Getting Started¶
Introduction¶
DPLib is a Python library that makes you able to write scripts that react on some in-game events
Installation¶
git clone https://github.com/mRokita/DPLib.git
cd DPLib
python3 -m pip install DPLib # Only python 3 is supported
First steps¶
This code is a basic example of how to use DPLib. It uses all the currently available events. You should definitely play around with it. Simply edit the third line to match your server’s config and run it!
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | # DPLib - Asynchronous bot framework for Digital Paint: Paintball 2 servers
# Copyright (C) 2017 Michał Rokita
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from dplib.server import Server
s = Server(hostname='127.0.0.1', port=27910, logfile=r'C:\Games\Paintball2\pball\qconsole27910.log', rcon_password='hello')
@s.event
def on_chat(nick, message):
print('Chat message. Nick: {0}, Message: {1}'.format(nick, message))
@s.event
def on_team_switched(nick, old_team, new_team):
print('Team switched. Nick: {0}, Old team: {1}, New team: {2}'.format(nick, old_team, new_team))
@s.event
def on_round_started():
print('Round started...')
@s.event
def on_elim(killer_nick, killer_weapon, victim_nick, victim_weapon):
print('Elimination. Killer\'s nick: {0}, Killer\'s weapon: {1}, Victim\'s nick: {2}, Victim\'s weapon: {3}'
.format(
killer_nick, killer_weapon, victim_nick, victim_weapon
))
@s.event
def on_respawn(team, nick):
print('Respawn. Nick: {0}, Team: {1}'.format(nick, team))
@s.event
def on_entrance(nick, build, addr):
print('Entrance. Nick: {0}, Build: {1}, Address: {2}'.format(
nick, build, addr
))
@s.event
def on_elim_teams_flag(team, nick, points):
print('Points for posession of eliminated teams flag. Team: {0}, Nick: {1}, Points: {2}'.format(
team, nick, points
))
@s.event
def on_flag_captured(team, nick, flag):
print('Flag captured. Team: {0}, Nick: {1}, Flag: {2}'.format(team, nick, flag))
@s.event
def on_game_end(score_blue, score_red, score_yellow, score_purple):
print('Game ended. Blue:{} Red:{} Yellow:{} Purple:{}'.format(
score_blue, score_red, score_yellow, score_purple
))
@s.event
def on_mapchange(mapname):
print('Map changed. Mapname: {}'.format(mapname))
@s.event
def on_namechange(old_nick, new_nick):
print('Name changed. Old nick: {} New nick: {}'.format(old_nick, new_nick))
print(s.get_status())
s.run()
|
Available event handlers¶
dplib.server.Server.on_message()
Waiting for future events¶
DPLib uses Python’s asyncio module so you can wait for incoming events without blocking the whole script.
Here’s a script that uses these ‘magic’ coroutines.
It’s a simple spawnkill protection system, it waits for 2 seconds after respawn for elimination event and when the newly respawned player gets killed within these 2 seconds, the spawnkiller gets a warning. After 4 spawnkills she/he gets kicked from the server.
Check out the 10th line.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | # DPLib - Asynchronous bot framework for Digital Paint: Paintball 2 servers
# Copyright (C) 2017 Michał Rokita
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from time import time
from dplib.server import Server
s = Server(hostname='127.0.0.1', port=27910, logfile=r'C:\Games\Paintball2\pball\qconsole27910.log', rcon_password='hello')
spawnkills = dict()
spawnkill_last_times = dict()
@s.event
def on_respawn(team, nick):
kill = yield from s.wait_for_elim(victim_nick=nick, timeout=2)
if not kill:
return
if not kill['killer_nick'] in spawnkills or time()-spawnkill_last_times[kill['killer_nick']] > 10:
spawnkills[kill['killer_nick']] = 0
spawnkills[kill['killer_nick']] += 1
spawnkill_last_times[kill['killer_nick']] = time()
s.say('{C}9%s, {C}A{U}STOP SPAWNKILLING{U}' % kill['killer_nick'])
if spawnkills[kill['killer_nick']] > 3:
s.kick(nick=kill['killer_nick'])
s.run()
|
Available coroutines¶
Examples¶
Map settings¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | # DPLib - Asynchronous bot framework for Digital Paint: Paintball 2 servers
# Copyright (C) 2017 Michał Rokita
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from asyncio import sleep
from dplib.server import Server
s = Server(hostname='127.0.0.1', port=27911, logfile=r'C:\Games\Paintball2\pball\qconsole27911.log', rcon_password='hello')
map_settings = {
'airtime': {
'command': 'set elim 10;set timelimit 10;',
'message': '{C}9Special settings for airtime {C}Aenabled'
},
'shazam33': {
'command': 'set elim 10;set timelimit 10;',
'message': '{C}9Special settings for shazam33 {C}Aenabled'
},
'default_settings': {
'command': 'set elim 20;set timelimit 20;',
'message': '{C}9No special settings for map {I}<mapname>{I}, using defaults'
}
}
@s.event
def on_mapchange(mapname):
if mapname not in map_settings:
settings = map_settings['default_settings']
else:
settings = map_settings[mapname]
command = settings.get('command', None)
message = settings.get('message', None)
if message:
message = mapname.join(message.split('<mapname>'))
if command:
for c in command.split(';'):
s.rcon(c)
if message:
yield from sleep(3)
s.say(message)
s.run()
|
Map elim script¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | # DPLib - Asynchronous bot framework for Digital Paint: Paintball 2 servers
# Copyright (C) 2017 Michał Rokita
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from dplib.server import Server
s = Server(hostname='127.0.0.1', port=27910, logfile=r'C:\Games\Paintball2\pball\qconsole27910.log', rcon_password='hello')
elim_active = False
@s.event
def on_chat(nick, message):
global elim_active
if message == '!map elim' and not elim_active:
elim_active = True
maps = ['beta/wobluda_fix', 'beta/daylight_b1', 'airtime']
s.say('{C}AType \'!elim <mapname>\' to eliminate a map.')
while len(maps) > 1:
s.say('{C}9Available maps: ' + ', '.join(maps))
msg = yield from s.wait_for_message(check=lambda n, m: m.startswith('!elim '))
mapname = msg['message'].split('!elim ')[1]
if mapname not in maps:
s.say('{C}9Invalid map.')
s.say('{C}9Available maps: ' + ', '.join(maps))
else:
maps.remove(mapname)
s.rcon('sv newmap '+maps[0])
elim_active = False
s.run()
|
Spawnkill message¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 | # DPLib - Asynchronous bot framework for Digital Paint: Paintball 2 servers
# Copyright (C) 2017 Michał Rokita
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from dplib.server import Server
s = Server(hostname='127.0.0.1', port=27910, logfile=r'C:\Games\Paintball2\pball\qconsole27910.log', rcon_password='hello')
@s.event
def on_respawn(team, nick):
kill = yield from s.wait_for_elim(victim_nick=nick, timeout=2)
if kill:
print(s.get_ingame_info(kill['killer_nick']).dplogin)
s.say('{C}9%s, {C}A{U}STOP SPAWNKILLING{U}' % kill['killer_nick'])
s.run()
|
Streak¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | # DPLib - Asynchronous bot framework for Digital Paint: Paintball 2 servers
# Copyright (C) 2017 Michał Rokita
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
from dplib.server import Server
s = Server(hostname='127.0.0.1', port=27910, logfile=r'C:\Games\Paintball2\pball\qconsole27910.log', rcon_password='hello')
@asyncio.coroutine
def streak(killer_nick):
for i in range(1, 3):
print(killer_nick, i)
yield from s.wait_for_elim(killer_nick=killer_nick)
@s.event
def on_elim(killer_nick, killer_weapon, victim_nick, victim_weapon):
try:
yield from asyncio.wait_for(streak(killer_nick), timeout=20)
s.say('{U}%s ZABUJCA!{U}' % killer_nick)
except asyncio.TimeoutError:
print('Timeout for '+ killer_nick)
s.run()
|
Server manager¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 | #!/usr/bin/python3
import asyncio
import shlex
import signal
import subprocess
import os
import sys
from multiprocessing import Process
from shlex import quote
from threading import Thread
import string
import random
from time import time, sleep
import pty
import aioftp
from _socket import timeout
from dplib.server import Server, ServerEvent
# -------------- BEGIN CONFIG SECTION ----------------
LAUNCH_TIMEOUT = 10
PAINTBALL_DIR = '/home/paintball/paintball2/'
SERVERS = {
'speed': {
'enabled': True,
'startup_cvars': {
'hostname': 'FALUBAZ/devol',
'public': 1,
'port': 27911,
'flagcapendsround': 0,
'elim': 10,
'rcon_password': ''.join(random.choice(string.ascii_letters) for i in range(10)),
},
'startup_commands': [
['setmaster', 'dplogin.com'],
['map', 'dodgeball'],
],
'event_rcons': {
ServerEvent.CHAT: 'say on_chat',
ServerEvent.ELIM: 'say on_elim',
ServerEvent.RESPAWN: 'say on_respawn',
ServerEvent.ENTRANCE: 'say Hello, {nick}',
ServerEvent.FLAG_CAPTURED: 'say on_flag_captured',
ServerEvent.ELIM_TEAMS_FLAG: 'say on_elim_teams_flag',
ServerEvent.ROUND_STARTED: 'say on_round_started',
ServerEvent.TEAM_SWITCHED: 'say on_team_switched',
ServerEvent.GAME_END: 'say on_game_end',
ServerEvent.MAPCHANGE: 'say on_mapchange',
ServerEvent.NAMECHANGE: 'say on_namechange',
},
'event_lambdas': {
ServerEvent.CHAT: lambda **kwargs: print(kwargs),
}
}
}
# -------------- END CONFIG SECTION ----------------
managed_servers = dict()
def get_map_download_path(mapname):
return os.path.join(PAINTBALL_DIR, 'pball', 'maps', mapname + '.bsp')
def get_map_server_path(mapname):
return os.path.join('pball', 'maps', mapname + '.bsp')
class ManagedServer(Server):
def __init__(self, server_id: str, config: dict):
self.config = config
self.server_id = server_id
self.pty_master = None
self.pty_slave = None
async def on_chat(self, nick, message):
if message.find('!map ') == 0:
mapname = message.split('!map ')[1:][0]
if os.path.exists(get_map_download_path(mapname)):
self.say("{C}9Map {C}A" + mapname + "{C}9 is already on the server, changing")
self.new_map(mapname)
else:
async with aioftp.ClientSession("ic3y.de", 21) as client:
if await client.exists(get_map_server_path(mapname)):
self.say(
"{C}9Downloading {C}A" + mapname + "{C}9 from ic3y.de...")
print(get_map_server_path(mapname))
await client.download(get_map_server_path(mapname), get_map_download_path(mapname), write_into=True)
self.say(
"{C}A" + mapname + "{C}9 has been downloaded successfully, changing map...")
self.new_map(mapname)
else:
self.say(
"{C}9 Map {C}A" + mapname + "{C}9 is not available at ic3y.de")
self.new_map(mapname)
@property
def running(self) -> bool:
return self.pid is not None
def start_event_service(self):
t = Thread(target=self.run, kwargs={'debug': True, 'make_secure': False})
t.start()
def get_event_handler(self, event_type):
if event_type in self.handlers:
async def handle(**kwargs):
if event_type in self.config['event_rcons']:
self.rcon(self.config['event_rcons'][event_type].format(
**kwargs))
if event_type in self.config['event_lambdas']:
self.config['event_lambdas'][event_type](**kwargs)
await getattr(self, self.handlers[event_type])(**kwargs)
return handle
else:
return super().get_event_handler(event_type)
def start_process(self):
self.kill_running_process()
if not self.config['enabled']:
print('[WARNING] Skipping server %s - "enabled" is False' %
self.server_id)
return
master, slave = os.openpty()
process = subprocess.Popen(shlex.split(self.get_run_command()),
stdout=slave,
stdin=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
preexec_fn=os.setsid,
shell=False)
self.pid = process.pid
start_time = time()
super(ManagedServer, self).__init__(
hostname='localhost',
port=self.config['startup_cvars']['port'],
rcon_password=self.config['startup_cvars']['rcon_password'],
pty_master=master
)
managed_servers[self.server_id] = self
port = None
self.make_secure()
self.start_event_service()
while not port and time() - start_time < LAUNCH_TIMEOUT:
try:
port = self.get_cvar('port')
except Exception as e:
pass
if not port:
print("[ERROR] Couldn't start server %s" % self.server_id)
self.kill()
else:
print("[SUCCESS] Server %s is running on 127.0.0.1:%s. PID: %d" %
(self.server_id, port, self.pid))
def get_run_command(self) -> str:
command = './paintball2 +set dedicated 1'
for v in self.config['startup_cvars']:
command += ' +set %s "%s"' % (quote(v), quote(str(self.config['startup_cvars'][v])))
for c in self.config['startup_commands']:
command += ' +' + ' '.join(quote(i) for i in c)
print(command)
return command
@property
def pid(self):
return self.__pid
@pid.setter
def pid(self, pid):
self.__pid = pid
if pid is None:
return
with open("%s.pid" % self.server_id, 'w+') as fo:
fo.write(str(self.__pid))
def kill(self):
self.stop_listening()
if self.pid:
try:
self.rcon("quit")
self.kill_own_process()
except timeout or ConnectionError:
if self.pid:
self.kill_own_process()
else:
self.kill_running_process()
print('[INFO] Killed server %s' % self.server_id)
self.pid = None
if self.server_id in managed_servers:
del managed_servers[self.server_id]
def kill_running_process(self):
file = "%s.pid" % self.server_id
if file not in os.listdir('.'):
return False
with open(file, "r") as fo:
pid = int(fo.read())
return self.kill_pid(pid)
def kill_own_process(self) -> bool:
return self.kill_pid(self.pid)
@staticmethod
def kill_pid(pid):
try:
os.killpg(pid, signal.SIGTERM)
print("Kill", pid)
return True
except ProcessLookupError:
print("Fail", pid)
return False
def run_servers():
for s in SERVERS:
managed_server = ManagedServer(server_id=s, config=SERVERS[s])
managed_server.start_process()
def kill_all():
for s in list(managed_servers.values()):
s.kill()
try:
if __name__ == '__main__':
os.chdir(PAINTBALL_DIR)
run_servers()
sleep(1)
print("Enter 'quit' to kill all servers")
wait = False
for s in list(managed_servers.values()):
if s.is_listening:
wait = True
break
if wait:
try:
while not input() == 'quit':
pass
kill_all()
except EOFError:
pass
except KeyboardInterrupt:
kill_all()
except Exception as e:
kill_all()
finally:
kill_all()
|