1
|
|
|
from abc import ABCMeta, abstractmethod |
2
|
|
|
|
3
|
|
|
|
4
|
|
|
from typing import List, Union |
|
|
|
|
5
|
|
|
from server.abc.base_game import InitMode |
|
|
|
|
6
|
|
|
|
7
|
|
|
|
8
|
|
|
class GpgNetServerProtocol(metaclass=ABCMeta): |
|
|
|
|
9
|
|
|
""" |
10
|
|
|
Defines an interface for the server side GPGNet protocol |
11
|
|
|
""" |
12
|
|
|
def send_CreateLobby(self, init_mode: InitMode, port: int, login: str, uid: int, natTraversalProvider: int): |
13
|
|
|
""" |
14
|
|
|
Tells the client to create a new LobbyComm instance and have it listen on the given port number |
15
|
|
|
:type init_mode: Whether to use ranked or ladder mode for the in game lobby |
16
|
|
|
:type port: The port number for the client to listen on |
17
|
|
|
:type login: The username of the player |
18
|
|
|
:type uid: The identifier of the player |
19
|
|
|
:type natTraversalProvider: A number representing the nat-traversal-provider, typically 1 |
20
|
|
|
""" |
21
|
|
|
self.send_gpgnet_message('CreateLobby', [int(init_mode.value), port, login, uid, natTraversalProvider]) |
22
|
|
|
|
23
|
|
|
def send_ConnectToPeer(self, address_and_port: str, player_name: str, player_uid: int): |
24
|
|
|
""" |
25
|
|
|
Tells a client that has a listening LobbyComm instance to connect to the given peer |
26
|
|
|
:param address_and_port: String of the form "adress:port" |
27
|
|
|
:param player_name: Remote player name |
28
|
|
|
:param player_uid: Remote player identifier |
29
|
|
|
""" |
30
|
|
|
self.send_gpgnet_message('ConnectToPeer', [address_and_port, player_name, player_uid]) |
31
|
|
|
|
32
|
|
|
def send_JoinGame(self, address_and_port: str, remote_player_name: str, remote_player_uid: int): |
33
|
|
|
""" |
34
|
|
|
Tells the game to join the given peer by address_and_port |
35
|
|
|
:param address_and_port: |
36
|
|
|
:param remote_player_name: |
37
|
|
|
:param remote_player_uid: |
38
|
|
|
""" |
39
|
|
|
self.send_gpgnet_message('JoinGame', [address_and_port, remote_player_name, remote_player_uid]) |
40
|
|
|
|
41
|
|
|
def send_HostGame(self, map): |
|
|
|
|
42
|
|
|
""" |
43
|
|
|
Tells the game to start listening for incoming connections as a host |
44
|
|
|
:param map: Which scenario to use |
45
|
|
|
""" |
46
|
|
|
self.send_gpgnet_message('HostGame', [str(map)]) |
47
|
|
|
|
48
|
|
|
def send_SendNatPacket(self, address_and_port: str, message: str): |
49
|
|
|
""" |
50
|
|
|
Instructs the game to send a nat-traversal UDP packet to the given remote address and port. |
51
|
|
|
|
52
|
|
|
The game will send the message verbatim as UDP-datagram prefixed with a \0x08 byte. |
53
|
|
|
:param address_and_port: |
54
|
|
|
:param message: |
55
|
|
|
""" |
56
|
|
|
self.send_gpgnet_message('SendNatPacket', [address_and_port, message]) |
57
|
|
|
|
58
|
|
|
def send_DisconnectFromPeer(self, id: int): |
|
|
|
|
59
|
|
|
""" |
60
|
|
|
Instructs the game to disconnect from the peer given by id |
61
|
|
|
|
62
|
|
|
:param id: |
63
|
|
|
:return: |
64
|
|
|
""" |
65
|
|
|
self.send_gpgnet_message('DisconnectFromPeer', [id]) |
66
|
|
|
|
67
|
|
|
|
68
|
|
|
def send_Ping(self): |
69
|
|
|
""" |
70
|
|
|
Heartbeat pinging used between the FAF client and server |
71
|
|
|
:return: |
72
|
|
|
""" |
73
|
|
|
self.send_gpgnet_message('ping', []) |
74
|
|
|
|
75
|
|
|
def send_gpgnet_message(self, command_id, arguments): |
76
|
|
|
message = {"command": command_id, "args": arguments} |
77
|
|
|
self.send_message(message) |
78
|
|
|
|
79
|
|
|
@abstractmethod |
80
|
|
|
def send_message(self, message): |
81
|
|
|
pass # pragma: no cover |
82
|
|
|
|
83
|
|
|
|
84
|
|
|
class GpgNetClientProtocol(metaclass=ABCMeta): |
|
|
|
|
85
|
|
|
def send_GameState(self, arguments: List[Union[int, str, bool]]) -> None: |
86
|
|
|
""" |
87
|
|
|
Sent by the client when the state of LobbyComm changes |
88
|
|
|
""" |
89
|
|
|
self.send_gpgnet_message('GameState', arguments) |
90
|
|
|
|
91
|
|
|
def send_ProcessNatPacket(self, arguments: List[Union[int, str, bool]]) -> None: |
92
|
|
|
""" |
93
|
|
|
Sent by the client when it received a nat packet |
94
|
|
|
:param arguments: |
95
|
|
|
:return: |
96
|
|
|
""" |
97
|
|
|
self.send_gpgnet_message('ProcessNatPacket', arguments) |
98
|
|
|
|
99
|
|
|
@abstractmethod |
100
|
|
|
def send_gpgnet_message(self, command_id, arguments: List[Union[int, str, bool]]) -> None: |
101
|
|
|
pass # pragma: no cover |
102
|
|
|
|
This can be caused by one of the following:
1. Missing Dependencies
This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands.
2. Missing __init__.py files
This error could also result from missing
__init__.py
files in your module folders. Make sure that you place one file in each sub-folder.