gsd
/
LinuxGSM
mirror of https://github.com/GameServerManagers/LinuxGSM
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Cleanup 

- PEP-8: spaces should be used for indentation
- "is" should be used for comparing with "None"
- corrected indentation of "self.connected = False"
pull/1128/head
Marvin Lehmann 9 years ago
parent
94de6e85f9
commit
e4c09f6f8f
1 changed files with 113 additions and 112 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 225
      lgsm/functions/gsquery.py

225
lgsm/functions/gsquery.py
View File

@ -9,122 +9,123 @@ import optparse
import socket import socket
import sys import sys
class GameServer: class GameServer:
def __init__( self, options, arguments ): def __init__(self, options, arguments):
self.option = options self.option = options
self.argument = arguments self.argument = arguments
# #
self.server_response_timeout = 5 self.server_response_timeout = 5
self.default_buffer_length = 1024 self.default_buffer_length = 1024
# #
if self.option.engine == 'avalanche': if self.option.engine == 'avalanche':
self.query_prompt_string = b'\xFE\xFD\x09\x10\x20\x30\x40' self.query_prompt_string = b'\xFE\xFD\x09\x10\x20\x30\x40'
elif self.option.engine == 'goldsource':
elif self.option.engine == 'goldsource': self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0' elif self.option.engine == 'idtech3':
elif self.option.engine == 'idtech3': self.query_prompt_string = b'\xff\xff\xff\xffgetstatus'
self.query_prompt_string = b'\xff\xff\xff\xffgetstatus' elif self.option.engine == 'quakelive':
elif self.option.engine == 'quakelive': self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0' elif self.option.engine == 'realvirtuality':
elif self.option.engine == 'realvirtuality': self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0' elif self.option.engine == 'refractor':
elif self.option.engine == 'refractor': self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0' elif self.option.engine == 'source':
elif self.option.engine == 'source': self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0' elif self.option.engine == 'spark':
elif self.option.engine == 'spark': self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0' elif self.option.engine == 'unity3d':
elif self.option.engine == 'unity3d': self.query_prompt_string = '\xFF\xFF\xFF\xFFTSource Engine Query\0'
self.query_prompt_string = '\xFF\xFF\xFF\xFFTSource Engine Query\0' elif self.option.engine == 'unreal':
elif self.option.engine == 'unreal': self.query_prompt_string = b'\x5C\x69\x6E\x66\x6F\x5C'
self.query_prompt_string = b'\x5C\x69\x6E\x66\x6F\x5C' elif self.option.engine == 'unreal2':
elif self.option.engine == 'unreal2': self.query_prompt_string = b'\x79\x00\x00\x00\x00'
self.query_prompt_string = b'\x79\x00\x00\x00\x00' self.connected = False
self.connected = False self.response = None
self.response = None self.sanity_checks()
self.sanity_checks()
def fatal_error(self, error_message, error_code=1):
sys.stderr.write('ERROR: ' + str(error_message) + '\n')
sys.exit(error_code)
def fatal_error( self, error_message, error_code=1 ): def exit_success(self, success_message=''):
sys.stderr.write( 'ERROR: ' + str(error_message) + '\n' ) sys.stdout.write('OK: ' + str(success_message) + '\n')
sys.exit( error_code ) sys.exit(0)
def exit_success( self, success_message='' ): def responding(self):
sys.stdout.write( 'OK: ' + str(success_message) + '\n' ) # Connect.
sys.exit( 0 ) connection = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
connection.settimeout(self.server_response_timeout)
try:
self.connected = connection.connect((self.option.address, int(self.option.port)))
except socket.timeout:
self.fatal_error('Request timed out', 1)
except:
self.fatal_error('Unable to connect', 1)
# Send.
connection.send(self.query_prompt_string)
# Receive.
try:
self.response = connection.recv(self.default_buffer_length)
except socket.error:
self.fatal_error('Unable to receive', 2)
connection.close()
# Response.
if self.response is None:
self.fatal_error('No response', 3)
if len(self.response) < 10:
sys.exit('Short response.', 3)
else:
self.exit_success(str(self.response))
def responding( self ): def sanity_checks(self):
# Connect. if not self.option.address:
connection = socket.socket( socket.AF_INET, socket.SOCK_DGRAM ) self.fatal_error('No IPv4 address supplied.', 4)
connection.settimeout( self.server_response_timeout ) if not self.option.port:
try: self.fatal_error('No port supplied.', 4)
self.connected = connection.connect( ( self.option.address, int(self.option.port) ) )
except socket.timeout:
self.fatal_error( 'Request timed out', 1 )
except:
self.fatal_error( 'Unable to connect', 1 )
# Send.
connection.send( self.query_prompt_string )
# Receive.
try:
self.response = connection.recv( self.default_buffer_length )
except socket.error:
self.fatal_error( 'Unable to receive', 2 )
connection.close()
# Response.
if self.response == None:
self.fatal_error( 'No response', 3 )
if len( self.response ) < 10 :
sys.exit( 'Short response.', 3 )
else:
self.exit_success( str( self.response ) )
def sanity_checks( self ):
if not self.option.address:
self.fatal_error( 'No IPv4 address supplied.', 4 )
if not self.option.port:
self.fatal_error( 'No port supplied.', 4 )
if __name__ == '__main__': if __name__ == '__main__':
parser = optparse.OptionParser( parser = optparse.OptionParser(
usage = 'usage: python %prog [options]', usage='usage: python %prog [options]',
version = '%prog 0.0.1' version='%prog 0.0.1'
) )
parser.add_option( parser.add_option(
'-a', '--address', '-a', '--address',
action = 'store', action='store',
dest = 'address', dest='address',
default = False, default=False,
help = 'The IPv4 address of the server.' help='The IPv4 address of the server.'
) )
parser.add_option( parser.add_option(
'-p', '--port', '-p', '--port',
action = 'store', action='store',
dest = 'port', dest='port',
default = False, default=False,
help = 'The IPv4 port of the server.' help='The IPv4 port of the server.'
) )
parser.add_option( parser.add_option(
'-e', '--engine', '-e', '--engine',
action = 'store', action='store',
dest = 'engine', dest='engine',
default = False, default=False,
help = 'Engine type: avalanche, goldsource, idtech3, realvirtuality, quakelive, refractor, spark, source, unity3d, unreal, unreal2.' help='Engine type: avalanche, goldsource, idtech3, realvirtuality, quakelive, refractor, spark, source, unity3d, unreal, unreal2.'
) )
parser.add_option( parser.add_option(
'-v', '--verbose', '-v', '--verbose',
action = 'store_true', action='store_true',
dest = 'verbose', dest='verbose',
default = False, default=False,
help = 'Display verbose output.' help='Display verbose output.'
) )
parser.add_option( parser.add_option(
'-d', '--debug', '-d', '--debug',
action = 'store_true', action='store_true',
dest = 'debug', dest='debug',
default = False, default=False,
help = 'Display debugging output.' help='Display debugging output.'
) )
options, arguments = parser.parse_args() options, arguments = parser.parse_args()
# #
server = GameServer( options, arguments ) server = GameServer(options, arguments)
server.responding() server.responding()

Write Preview
Loading…
Cancel
Save