tw_serverinfo.game_servers.GameServers.parse_extended_response() - Code Metrics - Inspection of "[CLEANUP] extract method for parsing the different..." - DaRealFreak/Teeworlds-ServerInfo - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6131fb...a89286 )

by Steffen

created 2018-10-10 10:12 UTC

GameServers.parse_extended_response() B

↳ Parent: tw_serverinfo.game_servers

Complexity

Conditions

Size

Total Lines	51
Code Lines	41

Duplication

Lines	12
Ratio	23.53 %

Code Coverage

Tests	2
CRAP Score	9.561

Importance

Changes

Metric	Value
cc	3
eloc	41
nop	2
dl	12
loc	51
ccs	2
cts	20
cp	0.1
crap	9.561
rs	8.896
c	0
b	0
f	0

How to fix Long Method

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

from tw_serverinfo import Network


class GameServers(object):
    SERVER_CHUNK_SIZE = 50
    SERVER_CHUNK_SLEEP_MS = 1000

    _game_servers = {}

    def fill_server_info(self, game_servers: dict) -> dict:
        """

        :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 key, game_server in self._game_servers.items():
            i += 1
            game_server = Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO'],
                                              server=game_server)
            game_server = Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO_64_LEGACY'],
                                              server=game_server)
            # update game server which got extended by the token
            self._game_servers[key] = 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
        return self._game_servers

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

        :type data: bytes
        :type server: dict
        :return:
        """
        game_server_key = '{ip:s}:{port:d}'.format(ip=server['ip'], port=server['port'])
        server = self._game_servers[game_server_key]

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

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

    @staticmethod
    def parse_vanilla_response(slots: deque, server: dict) -> dict:
        """Parse the default response of the vanilla client

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

        players = server['players']
        while len(slots) >= 5:
            # no idea what this is, always empty
            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'))
            })

        return {
            'ip': server['ip'],
            'port': server['port'],
            'type': server['type'],
            'server_type': 'vanilla',
            'version': version,
            'name': name,
            'map_name': map_name,
            'game_type': game_type,
            'flags': flags,
            'num_players': num_players,
            'max_players': max_players,
            'num_clients': num_clients,
            'max_clients': max_clients,
            'players': players
        }

    @staticmethod
    def parse_64_legacy_response(slots: deque, server: dict) -> dict:
        """Parse the 64 slot legacy response

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

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

        while len(slots) >= 5:

            player = {
                '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'))
            }
            if player not in players:
                players.append(player)

        return {
            'ip': server['ip'],
            'port': server['port'],
            'type': server['type'],
            'server_type': '64_legacy',
            'version': version,
            'name': name,
            'map_name': map_name,
            'game_type': game_type,
            'flags': flags,
            'num_players': num_players,
            'max_players': max_players,
            'num_clients': num_clients,
            'max_clients': max_clients,
            'players': players
        }

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

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

        players = server['players']
        while len(slots) >= 6:

            # no idea what this is, always empty
            slots.popleft()
            player = {
                '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'))
            }
            if player not in players:
                players.append(player)

        return {
            'ip': server['ip'],
            'port': server['port'],
            'type': server['type'],
            'server_type': 'ext',
            'version': version,
            'name': name,
            'map_name': map_name,
            'map_crc': map_crc,
            'map_size': map_size,
            'game_type': game_type,
            'flags': flags,
            'num_players': num_players,
            'max_players': max_players,
            'num_clients': num_clients,
            'max_clients': max_clients,
            'players': players
        }


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

DaRealFreak / Teeworlds-ServerInfo

Push — master ( 6131fb...a89286 )

GameServers.parse_extended_response() B

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like