tw_serverinfo.game_servers.GameServers.parse_vanilla_response() - Code Metrics - Inspection of "[TASK] set game server response on parsing the pac..." - DaRealFreak/Teeworlds-ServerInfo - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( fe5519...789e01 )

by Steffen

created 2018-10-12 12:02 UTC

GameServers.parse_vanilla_response() A

↳ Parent: tw_serverinfo.game_servers

Complexity

Conditions

Size

Total Lines	27
Code Lines	19

Duplication

Lines	27
Ratio	100 %

Code Coverage

Tests	2
CRAP Score	4.5185

Importance

Changes

Metric	Value
cc	2
eloc	19
nop	2
dl	27
loc	27
ccs	2
cts	14
cp	0.1429
crap	4.5185
rs	9.45
c	0
b	0
f	0

#!/usr/bin/python
# -*- coding: utf-8 -*-
import logging
import socket
from collections import deque
from time import sleep

from tw_serverinfo import Network
from tw_serverinfo.models.game_server import GameServer


class GameServers(object):
    """GameServers Class containing the logic to request and parse responses from the game servers"""
    SERVER_CHUNK_SIZE = 50
    SERVER_CHUNK_SLEEP_MS = 1000

    _game_servers = []

    def fill_server_info(self, game_servers: list) -> None:
        """

        :param game_servers:
        :return:
        """
        self._game_servers = game_servers
        # create an udp protocol socket
        sock = socket.socket(family=Network.PROTOCOL_FAMILY, type=socket.SOCK_DGRAM)
        # set the socket to non blocking to allow parallel requests
        sock.setblocking(False)

        i = 0
        for game_server in self._game_servers:  # type: GameServer
            i += 1
            Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO'], server=game_server)
            Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO_64_LEGACY'], server=game_server)

            if i % self.SERVER_CHUNK_SIZE or i >= len(self._game_servers):
                duration_without_response = Network.CONNECTION_SLEEP_DURATION
                sleep(Network.CONNECTION_SLEEP_DURATION / 1000.0)

                while True:
                    if not Network.receive_packet(sock, self._game_servers, self.process_packet):
                        if duration_without_response > Network.CONNECTION_TIMEOUT:
                            # we didn't receive any packets in time and cancel the connection here
                            sock.close()
                            break
                        else:
                            # increase the measured duration without a response and sleep for the set duration
                            duration_without_response += Network.CONNECTION_SLEEP_DURATION
                            sleep(Network.CONNECTION_SLEEP_DURATION / 1000.0)
                    else:
                        # if we got a response reset the duration in case we receive multiple packets
                        duration_without_response = 0

    def process_packet(self, data: bytes, server: GameServer) -> None:
        """Process packet function for
         - SERVERBROWSE_COUNT
         - SERVERBROWSE_LIST
        packets

        :type data: bytes
        :type server: GameServer
        :return:
        """
        slots = deque(data[14:].split(b"\x00"))
        token = int(slots.popleft().decode('utf-8'))
        # ToDo: token validation

        server.response = True

        if data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO']:
            # vanilla
            self.parse_vanilla_response(slots, server)
        elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_64_LEGACY']:
            # 64 legacy
            self.parse_64_legacy_response(slots, server)
        elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED']:
            # extended response, current default of DDNet
            self.parse_extended_response(slots, server)
        elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED_MORE']:
            logging.log(logging.DEBUG, 'no idea what to expect here, never got useful data')

    @staticmethod

    def parse_vanilla_response(slots: deque, server: GameServer) -> None:
        """Parse the default response of the vanilla client

        :param slots:
        :param server:
        :return:
        """
        server.server_type = 'vanilla'
        server.version = slots.popleft().decode('utf-8')
        server.name = slots.popleft().decode('utf-8')
        server.map_name = slots.popleft().decode('utf-8')
        server.game_type = slots.popleft().decode('utf-8')
        server.flags = int(slots.popleft().decode('utf-8'))
        server.num_players = int(slots.popleft().decode('utf-8'))
        server.max_players = int(slots.popleft().decode('utf-8'))
        server.num_clients = int(slots.popleft().decode('utf-8'))
        server.max_clients = int(slots.popleft().decode('utf-8'))

        while len(slots) >= 5:
            # no idea what this is, always empty
            server.players.append({
                'name': slots.popleft(),
                'clan': slots.popleft(),
                'country': int(slots.popleft().decode('utf-8')),
                'score': int(slots.popleft().decode('utf-8')),
                'ingame': int(slots.popleft().decode('utf-8'))
            })

    @staticmethod

    def parse_64_legacy_response(slots: deque, server: GameServer) -> None:
        """Parse the 64 slot legacy response

        :param slots:
        :param server:
        :return:
        """
        server.server_type = '64_legacy'
        server.version = slots.popleft().decode('utf-8')
        server.name = slots.popleft().decode('utf-8')
        server.map_name = slots.popleft().decode('utf-8')
        server.game_type = slots.popleft().decode('utf-8')
        server.flags = int(slots.popleft().decode('utf-8'))
        server.num_players = int(slots.popleft().decode('utf-8'))
        server.max_players = int(slots.popleft().decode('utf-8'))
        server.num_clients = int(slots.popleft().decode('utf-8'))
        server.max_clients = int(slots.popleft().decode('utf-8'))

        if slots[0] == b'':
            # no idea what this is, always empty
            slots.popleft()

        while len(slots) >= 5:
            server.players.append({
                'name': slots.popleft(),
                'clan': slots.popleft(),
                'country': int(slots.popleft().decode('utf-8')),
                'score': int(slots.popleft().decode('utf-8')),
                'ingame': int(slots.popleft().decode('utf-8'))
            })

    @staticmethod
    def parse_extended_response(slots: deque, server: GameServer) -> None:
        """Parse the extended server info response(default for DDNet)

        :param slots:
        :param server:
        :return:
        """
        server.server_type = 'ext'
        server.version = slots.popleft().decode('utf-8')
        server.name = slots.popleft().decode('utf-8')
        server.map_name = slots.popleft().decode('utf-8')
        server.map_crc = int(slots.popleft().decode('utf-8'))
        server.map_size = int(slots.popleft().decode('utf-8'))
        server.game_type = slots.popleft().decode('utf-8')
        server.flags = int(slots.popleft().decode('utf-8'))
        server.num_players = int(slots.popleft().decode('utf-8'))
        server.max_players = int(slots.popleft().decode('utf-8'))
        server.num_clients = int(slots.popleft().decode('utf-8'))
        server.max_clients = int(slots.popleft().decode('utf-8'))

        while len(slots) >= 6:
            # no idea what this is, always empty
            slots.popleft()
            server.players.append({
                'name': slots.popleft(),
                'clan': slots.popleft(),
                'country': int(slots.popleft().decode('utf-8')),
                'score': int(slots.popleft().decode('utf-8')),
                'ingame': int(slots.popleft().decode('utf-8'))
            })


1			#!/usr/bin/python
2			# -- coding: utf-8 --
3	1		import logging
4	1		import socket
5	1		from collections import deque
6	1		from time import sleep
7
8	1		from tw_serverinfo import Network
9	1		from tw_serverinfo.models.game_server import GameServer
10
11
12	1		class GameServers(object):
13			"""GameServers Class containing the logic to request and parse responses from the game servers"""
14	1		SERVER_CHUNK_SIZE = 50
15	1		SERVER_CHUNK_SLEEP_MS = 1000
16
17	1		_game_servers = []
18
19	1		def fill_server_info(self, game_servers: list) -> None:
20			"""
21
22			:param game_servers:
23			:return:
24			"""
25	1		self._game_servers = game_servers
26			# create an udp protocol socket
27	1		sock = socket.socket(family=Network.PROTOCOL_FAMILY, type=socket.SOCK_DGRAM)
28			# set the socket to non blocking to allow parallel requests
29	1		sock.setblocking(False)
30
31	1		i = 0
32	1		for game_server in self._game_servers: # type: GameServer
33	1		i += 1
34	1		Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO'], server=game_server)
35	1		Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO_64_LEGACY'], server=game_server)
36
37	1		if i % self.SERVER_CHUNK_SIZE or i >= len(self._game_servers):
38	1		duration_without_response = Network.CONNECTION_SLEEP_DURATION
39	1		sleep(Network.CONNECTION_SLEEP_DURATION / 1000.0)
40
41	1		while True:
42	1		if not Network.receive_packet(sock, self._game_servers, self.process_packet):
43	1		if duration_without_response > Network.CONNECTION_TIMEOUT:
44			# we didn't receive any packets in time and cancel the connection here
45	1		sock.close()
46	1		break
47			else:
48			# increase the measured duration without a response and sleep for the set duration
49	1		duration_without_response += Network.CONNECTION_SLEEP_DURATION
50	1		sleep(Network.CONNECTION_SLEEP_DURATION / 1000.0)
51			else:
52			# if we got a response reset the duration in case we receive multiple packets
53	1		duration_without_response = 0
54
55	1		def process_packet(self, data: bytes, server: GameServer) -> None:
56			"""Process packet function for
57			- SERVERBROWSE_COUNT
58			- SERVERBROWSE_LIST
59			packets
60
61			:type data: bytes
62			:type server: GameServer
63			:return:
64			"""
65	1		slots = deque(data[14:].split(b"\x00"))
66	1		token = int(slots.popleft().decode('utf-8'))
67			# ToDo: token validation
68
69	1		server.response = True
70
71	1		if data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO']:
72			# vanilla
73			self.parse_vanilla_response(slots, server)
74	1		elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_64_LEGACY']:
75			# 64 legacy
76	1		self.parse_64_legacy_response(slots, server)
77	1		elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED']:
78			# extended response, current default of DDNet
79	1		self.parse_extended_response(slots, server)
80			elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED_MORE']:
81			logging.log(logging.DEBUG, 'no idea what to expect here, never got useful data')
82
83	1	View Code Duplication	@staticmethod
			0 ignored issues – show Duplication introduced 2018-10-12 10:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
84	1		def parse_vanilla_response(slots: deque, server: GameServer) -> None:
85			"""Parse the default response of the vanilla client
86
87			:param slots:
88			:param server:
89			:return:
90			"""
91			server.server_type = 'vanilla'
92			server.version = slots.popleft().decode('utf-8')
93			server.name = slots.popleft().decode('utf-8')
94			server.map_name = slots.popleft().decode('utf-8')
95			server.game_type = slots.popleft().decode('utf-8')
96			server.flags = int(slots.popleft().decode('utf-8'))
97			server.num_players = int(slots.popleft().decode('utf-8'))
98			server.max_players = int(slots.popleft().decode('utf-8'))
99			server.num_clients = int(slots.popleft().decode('utf-8'))
100			server.max_clients = int(slots.popleft().decode('utf-8'))
101
102			while len(slots) >= 5:
103			# no idea what this is, always empty
104			server.players.append({
105			'name': slots.popleft(),
106			'clan': slots.popleft(),
107			'country': int(slots.popleft().decode('utf-8')),
108			'score': int(slots.popleft().decode('utf-8')),
109			'ingame': int(slots.popleft().decode('utf-8'))
110			})
111
112	1	View Code Duplication	@staticmethod
			0 ignored issues – show Duplication introduced 2018-10-12 10:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
113	1		def parse_64_legacy_response(slots: deque, server: GameServer) -> None:
114			"""Parse the 64 slot legacy response
115
116			:param slots:
117			:param server:
118			:return:
119			"""
120	1		server.server_type = '64_legacy'
121	1		server.version = slots.popleft().decode('utf-8')
122	1		server.name = slots.popleft().decode('utf-8')
123	1		server.map_name = slots.popleft().decode('utf-8')
124	1		server.game_type = slots.popleft().decode('utf-8')
125	1		server.flags = int(slots.popleft().decode('utf-8'))
126	1		server.num_players = int(slots.popleft().decode('utf-8'))
127	1		server.max_players = int(slots.popleft().decode('utf-8'))
128	1		server.num_clients = int(slots.popleft().decode('utf-8'))
129	1		server.max_clients = int(slots.popleft().decode('utf-8'))
130
131	1		if slots[0] == b'':
132			# no idea what this is, always empty
133	1		slots.popleft()
134
135	1		while len(slots) >= 5:
136	1		server.players.append({
137			'name': slots.popleft(),
138			'clan': slots.popleft(),
139			'country': int(slots.popleft().decode('utf-8')),
140			'score': int(slots.popleft().decode('utf-8')),
141			'ingame': int(slots.popleft().decode('utf-8'))
142			})
143
144	1		@staticmethod
145	1		def parse_extended_response(slots: deque, server: GameServer) -> None:
146			"""Parse the extended server info response(default for DDNet)
147
148			:param slots:
149			:param server:
150			:return:
151			"""
152	1		server.server_type = 'ext'
153	1		server.version = slots.popleft().decode('utf-8')
154	1		server.name = slots.popleft().decode('utf-8')
155	1		server.map_name = slots.popleft().decode('utf-8')
156	1		server.map_crc = int(slots.popleft().decode('utf-8'))
157	1		server.map_size = int(slots.popleft().decode('utf-8'))
158	1		server.game_type = slots.popleft().decode('utf-8')
159	1		server.flags = int(slots.popleft().decode('utf-8'))
160	1		server.num_players = int(slots.popleft().decode('utf-8'))
161	1		server.max_players = int(slots.popleft().decode('utf-8'))
162	1		server.num_clients = int(slots.popleft().decode('utf-8'))
163	1		server.max_clients = int(slots.popleft().decode('utf-8'))
164
165	1		while len(slots) >= 6:
166			# no idea what this is, always empty
167	1		slots.popleft()
168	1		server.players.append({
169			'name': slots.popleft(),
170			'clan': slots.popleft(),
171			'country': int(slots.popleft().decode('utf-8')),
172			'score': int(slots.popleft().decode('utf-8')),
173			'ingame': int(slots.popleft().decode('utf-8'))
174			})
175

DaRealFreak / Teeworlds-ServerInfo

Push — master ( fe5519...789e01 )

GameServers.parse_vanilla_response() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like