tw_serverinfo.game_servers.GameServers.process_packet() - Code Metrics - Inspection of "[DOC] use fancy new flat square badges" - DaRealFreak/Teeworlds-ServerInfo - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ae8a23...ea2adf )

by Steffen

created 2018-12-05 17:00 UTC

GameServers.process_packet() B

↳ Parent: tw_serverinfo.game_servers

Complexity

Conditions

Size

Total Lines	31
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	6.0852

Importance

Changes

Metric	Value
cc	6
eloc	15
nop	3
dl	0
loc	31
ccs	13
cts	15
cp	0.8667
crap	6.0852
rs	8.6666
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 typing import List

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


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[GameServer]) -> None:
        """Send the SERVERBROWSE_GETINFO and SERVERBROWSE_GETINFO_64_LEGACY packet to each game server in the
        passed game_servers list, parse the response and update the GameServer objects in the game_servers argument

        :type game_servers: list
        :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'], extra_data=b'xe',
                                server=game_server)
            Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO_64_LEGACY'], extra_data=b'xe',
                                server=game_server)

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

                while duration_without_response < Network.CONNECTION_TIMEOUT:
                    if not Network.receive_packet(sock, self._game_servers, self.process_packet):
                        # 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

        # close the socket after checking all servers
        sock.close()

    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:
        """
        server.response = True
        slots = deque(data[14:].split(b"\x00"))

        # validate the server token
        token = int(slots.popleft().decode('utf-8'))
        if int.from_bytes(server.token, byteorder='big') != token & 0xff:
            logging.log(logging.INFO, 'token validation failed for GameServer response. GameServer: ' + str(server))
            return

        if data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO']:
            # vanilla (inf3) packet
            self.parse_vanilla_response(slots, server)
        elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_64_LEGACY']:
            # 64 legacy (dtsf) packet
            self.parse_64_legacy_response(slots, server)
        elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED']:
            # extended response (iext) packet, current default of DDNet
            self.parse_extended_response(slots, server, token)
        elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED_MORE']:
            # extended response (iex+) packet
            self.parse_extended_more_response(slots, server)

    @staticmethod

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

        :type slots: deque
        :type server: GameServer
        :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:
            server.append_player(Player(
                name=slots.popleft().decode('utf-8'),
                clan=slots.popleft().decode('utf-8'),
                country=int(slots.popleft().decode('utf-8')),
                score=int(slots.popleft().decode('utf-8')),
                ingame=bool(int(slots.popleft().decode('utf-8')))
            ))

    @staticmethod

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

        :type slots: deque
        :type server: GameServer
        :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.append_player(Player(
                name=slots.popleft().decode('utf-8'),
                clan=slots.popleft().decode('utf-8'),
                country=int(slots.popleft().decode('utf-8')),
                score=int(slots.popleft().decode('utf-8')),
                ingame=bool(int(slots.popleft().decode('utf-8')))
            ))

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

        :type slots: deque
        :type server: GameServer
        :type token: int
        :return:
        """
        if (token & 0xffff00) >> 8 != (server.request_token[0] << 8) + server.request_token[1]:
            # additional server token validation failed
            logging.log(logging.INFO, 'request token validation failed for GameServer response. GameServer: '
                        + str(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.append_player(Player(
                name=slots.popleft().decode('utf-8'),
                clan=slots.popleft().decode('utf-8'),
                country=int(slots.popleft().decode('utf-8')),
                score=int(slots.popleft().decode('utf-8')),
                ingame=bool(int(slots.popleft().decode('utf-8')))
            ))

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

        :type slots: deque
        :type server: GameServer
        :return:
        """
        slots.popleft()

        server.server_type = 'ext+'
        while len(slots) >= 6:

            # no idea what this is, always empty
            slots.popleft()
            server.append_player(Player(
                name=slots.popleft().decode('utf-8'),
                clan=slots.popleft().decode('utf-8'),
                country=int(slots.popleft().decode('utf-8')),
                score=int(slots.popleft().decode('utf-8')),
                ingame=bool(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	1		from typing import List
8
9	1		from tw_serverinfo import Network
10	1		from tw_serverinfo.models.game_server import GameServer
11	1		from tw_serverinfo.models.player import Player
12
13
14	1		class GameServers(object):
15			"""GameServers Class containing the logic to request and parse responses from the game servers"""
16	1		SERVER_CHUNK_SIZE = 50
17	1		SERVER_CHUNK_SLEEP_MS = 1000
18
19	1		_game_servers = []
20
21	1		def fill_server_info(self, game_servers: List[GameServer]) -> None:
22			"""Send the SERVERBROWSE_GETINFO and SERVERBROWSE_GETINFO_64_LEGACY packet to each game server in the
23			passed game_servers list, parse the response and update the GameServer objects in the game_servers argument
24
25			:type game_servers: list
26			:return:
27			"""
28	1		self._game_servers = game_servers
29			# create an udp protocol socket
30	1		sock = socket.socket(family=Network.PROTOCOL_FAMILY, type=socket.SOCK_DGRAM)
31			# set the socket to non blocking to allow parallel requests
32	1		sock.setblocking(False)
33
34	1		i = 0
35	1		for game_server in self._game_servers: # type: GameServer
36	1		i += 1
37	1		Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO'], extra_data=b'xe',
38			server=game_server)
39	1		Network.send_packet(sock=sock, data=Network.PACKETS['SERVERBROWSE_GETINFO_64_LEGACY'], extra_data=b'xe',
40			server=game_server)
41
42	1		if i % self.SERVER_CHUNK_SIZE == 0 or i >= len(self._game_servers):
43	1		duration_without_response = Network.CONNECTION_SLEEP_DURATION
44	1		sleep(Network.CONNECTION_SLEEP_DURATION / 1000.0)
45
46	1		while duration_without_response < Network.CONNECTION_TIMEOUT:
47	1		if not Network.receive_packet(sock, self._game_servers, self.process_packet):
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			# close the socket after checking all servers
56	1		sock.close()
57
58	1		def process_packet(self, data: bytes, server: GameServer) -> None:
59			"""Process packet function for
60			- SERVERBROWSE_COUNT
61			- SERVERBROWSE_LIST
62			packets
63
64			:type data: bytes
65			:type server: GameServer
66			:return:
67			"""
68	1		server.response = True
69	1		slots = deque(data[14:].split(b"\x00"))
70
71			# validate the server token
72	1		token = int(slots.popleft().decode('utf-8'))
73	1		if int.from_bytes(server.token, byteorder='big') != token & 0xff:
74			logging.log(logging.INFO, 'token validation failed for GameServer response. GameServer: ' + str(server))
75			return
76
77	1		if data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO']:
78			# vanilla (inf3) packet
79	1		self.parse_vanilla_response(slots, server)
80	1		elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_64_LEGACY']:
81			# 64 legacy (dtsf) packet
82	1		self.parse_64_legacy_response(slots, server)
83	1		elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED']:
84			# extended response (iext) packet, current default of DDNet
85	1		self.parse_extended_response(slots, server, token)
86	1		elif data[6:6 + 8] == Network.PACKETS['SERVERBROWSE_INFO_EXTENDED_MORE']:
87			# extended response (iex+) packet
88	1		self.parse_extended_more_response(slots, server)
89
90	1	View Code Duplication	@staticmethod
			0 ignored issues – show Duplication introduced 2018-10-13 22:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
91	1		def parse_vanilla_response(slots: deque, server: GameServer) -> None:
92			"""Parse the default response of the vanilla client
93
94			:type slots: deque
95			:type server: GameServer
96			:return:
97			"""
98	1		server.server_type = 'vanilla'
99	1		server.version = slots.popleft().decode('utf-8')
100	1		server.name = slots.popleft().decode('utf-8')
101	1		server.map_name = slots.popleft().decode('utf-8')
102	1		server.game_type = slots.popleft().decode('utf-8')
103	1		server.flags = int(slots.popleft().decode('utf-8'))
104	1		server.num_players = int(slots.popleft().decode('utf-8'))
105	1		server.max_players = int(slots.popleft().decode('utf-8'))
106	1		server.num_clients = int(slots.popleft().decode('utf-8'))
107	1		server.max_clients = int(slots.popleft().decode('utf-8'))
108
109	1		while len(slots) >= 5:
110	1		server.append_player(Player(
111			name=slots.popleft().decode('utf-8'),
112			clan=slots.popleft().decode('utf-8'),
113			country=int(slots.popleft().decode('utf-8')),
114			score=int(slots.popleft().decode('utf-8')),
115			ingame=bool(int(slots.popleft().decode('utf-8')))
116			))
117
118	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...
119	1		def parse_64_legacy_response(slots: deque, server: GameServer) -> None:
120			"""Parse the 64 slot legacy response
121
122			:type slots: deque
123			:type server: GameServer
124			:return:
125			"""
126	1		server.server_type = '64_legacy'
127	1		server.version = slots.popleft().decode('utf-8')
128	1		server.name = slots.popleft().decode('utf-8')
129	1		server.map_name = slots.popleft().decode('utf-8')
130	1		server.game_type = slots.popleft().decode('utf-8')
131	1		server.flags = int(slots.popleft().decode('utf-8'))
132	1		server.num_players = int(slots.popleft().decode('utf-8'))
133	1		server.max_players = int(slots.popleft().decode('utf-8'))
134	1		server.num_clients = int(slots.popleft().decode('utf-8'))
135	1		server.max_clients = int(slots.popleft().decode('utf-8'))
136
137	1		if slots[0] == b'':
138			# no idea what this is, always empty
139	1		slots.popleft()
140
141	1		while len(slots) >= 5:
142	1		server.append_player(Player(
143			name=slots.popleft().decode('utf-8'),
144			clan=slots.popleft().decode('utf-8'),
145			country=int(slots.popleft().decode('utf-8')),
146			score=int(slots.popleft().decode('utf-8')),
147			ingame=bool(int(slots.popleft().decode('utf-8')))
148			))
149
150	1		@staticmethod
151	1		def parse_extended_response(slots: deque, server: GameServer, token: int) -> None:
152			"""Parse the extended server info response(default for DDNet)
153
154			:type slots: deque
155			:type server: GameServer
156			:type token: int
157			:return:
158			"""
159	1		if (token & 0xffff00) >> 8 != (server.request_token[0] << 8) + server.request_token[1]:
160			# additional server token validation failed
161			logging.log(logging.INFO, 'request token validation failed for GameServer response. GameServer: '
162			+ str(server))
163			return
164
165	1		server.server_type = 'ext'
166	1		server.version = slots.popleft().decode('utf-8')
167	1		server.name = slots.popleft().decode('utf-8')
168	1		server.map_name = slots.popleft().decode('utf-8')
169	1		server.map_crc = int(slots.popleft().decode('utf-8'))
170	1		server.map_size = int(slots.popleft().decode('utf-8'))
171	1		server.game_type = slots.popleft().decode('utf-8')
172	1		server.flags = int(slots.popleft().decode('utf-8'))
173	1		server.num_players = int(slots.popleft().decode('utf-8'))
174	1		server.max_players = int(slots.popleft().decode('utf-8'))
175	1		server.num_clients = int(slots.popleft().decode('utf-8'))
176	1		server.max_clients = int(slots.popleft().decode('utf-8'))
177
178	1	View Code Duplication	while len(slots) >= 6:
			0 ignored issues – show Duplication introduced 2018-10-13 22:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
179			# no idea what this is, always empty
180	1		slots.popleft()
181	1		server.append_player(Player(
182			name=slots.popleft().decode('utf-8'),
183			clan=slots.popleft().decode('utf-8'),
184			country=int(slots.popleft().decode('utf-8')),
185			score=int(slots.popleft().decode('utf-8')),
186			ingame=bool(int(slots.popleft().decode('utf-8')))
187			))
188
189	1		@staticmethod
190	1		def parse_extended_more_response(slots: deque, server: GameServer) -> None:
191			"""Parse the extended server info response(default for DDNet)
192
193			:type slots: deque
194			:type server: GameServer
195			:return:
196			"""
197	1		slots.popleft()
198
199	1		server.server_type = 'ext+'
200	1	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...
201			# no idea what this is, always empty
202	1		slots.popleft()
203	1		server.append_player(Player(
204			name=slots.popleft().decode('utf-8'),
205			clan=slots.popleft().decode('utf-8'),
206			country=int(slots.popleft().decode('utf-8')),
207			score=int(slots.popleft().decode('utf-8')),
208			ingame=bool(int(slots.popleft().decode('utf-8')))
209			))
210

DaRealFreak / Teeworlds-ServerInfo

Push — master ( ae8a23...ea2adf )

GameServers.process_packet() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like