implemented checking for network connections in gui

This commit is contained in:
Nicholas Dyer 2022-12-21 12:30:10 -05:00
parent 5c1a61f25c
commit 99feadc7e2
2 changed files with 163 additions and 5 deletions

View File

@ -0,0 +1,91 @@
"""
Networking client to connect to a server and find server hosts
"""
import ipaddress
import random
import psutil
import socket
SERVER_PORT = 29987
SERVER_BUFFER = 1024
SCAN_TIMEOUT = 3
class InterfaceInfo:
"""
Class used for storing interface information to make it easier to use
"""
def __init__(self, interface_name, network):
self.interface_name = interface_name
self.network = network
def __str__(self):
return 'Interface:{0}, Network:{1}'.format(self.interface_name, self.network)
def __repr__(self):
return '(Interface:{0}, Network:{1})'.format(self.interface_name, self.network)
def request_server_info(server_ip, server_port):
"""
Request server info from a PyPong server given an IP and a port
:param server_ip: Server IP to request info from
:param server_port: Port to connect to server on
:return response given by server (or lack thereof):
"""
request = 'PYPONGREQ;SVRINFO'
udp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_client_socket.settimeout(SCAN_TIMEOUT + (SCAN_TIMEOUT * random.random()))
try:
udp_client_socket.sendto(str.encode(request, 'UTF-8'), (server_ip, server_port))
except OSError:
pass
try:
server_response = udp_client_socket.recvfrom(SERVER_BUFFER)[0].decode('UTF-8')
except TimeoutError:
server_response = '{0};TIMEOUT'.format(server_ip)
except UnicodeDecodeError:
server_response = '{0};MANGLED_RESPONSE'.format(server_ip)
udp_client_socket.close()
return server_response
def get_interface_info():
raw_interface_data = psutil.net_if_addrs()
interface_data = []
# Extract needed information from psutil and put it into a InterfaceInfo object
for interface in raw_interface_data:
address = (raw_interface_data.get(interface)[0]).address
cidr_suffix = ipaddress.IPv4Network('{0}/{1}'.format(address, raw_interface_data.get(interface)[0].netmask),
strict=False).prefixlen
if ':' not in address and address != '127.0.0.1':
interface_data.append(InterfaceInfo(interface, '{0}/{1}'.format(address, cidr_suffix)))
return interface_data
def get_ips_from_network(network):
network_split = network.split('/')
address = network_split[0]
cidr = int(network_split[1])
all_addresses = []
test_cidr = 32
while test_cidr >= cidr:
current_addresses = [str(ip) for ip in ipaddress.IPv4Network('{0}/{1}'.format(address, test_cidr),
strict=False)]
for test_address in current_addresses:
if test_address not in all_addresses:
all_addresses.append(test_address)
test_cidr -= 1
print(test_cidr)
return all_addresses
def check_ip(ip_address):
response = request_server_info(ip_address, SERVER_PORT)
return response

View File

@ -2,11 +2,15 @@
Starting launcher of the game. This is where you host new games or join other ones.
"""
import tkinter as tk
import concurrent.futures
import pypong.networking.client as client
global running
global listbox
def main():
executor = concurrent.futures.ThreadPoolExecutor(max_workers=512)
"""
Runs the game launcher.
"""
@ -14,7 +18,7 @@ def main():
main_window.title('PynPong Launcher')
main_window.resizable(width=False, height=False)
main_window.geometry('500x250')
main_window.geometry('500x300')
def quit_launcher():
"""
@ -22,20 +26,77 @@ def main():
"""
global running
running = False
executor.shutdown(wait=False, cancel_futures=True)
def get_games():
def check_game(*game_ip):
game_ip = ''.join(game_ip)
result = client.check_ip(game_ip)
return result
def get_games(network):
"""
Refresh the games list.
"""
results = []
listbox.delete(0, tk.END)
listbox.insert(0, 'Searching for games...')
# Networking code will go here eventually
listbox.insert(0, ' Building IP range...')
main_window.update()
possible_ips = client.get_ips_from_network(network.get().split(' ')[0])
listbox.delete(0, tk.END)
listbox.insert(0, 'this might take some time...')
listbox.insert(0, 'Refreshing server listing,')
main_window.update()
# possible_ips = ['192.168.12.19']
futures = []
for ip in possible_ips:
if ip == '192.168.12.19':
print(ip)
future = executor.submit(check_game, ip)
futures.append(future)
game_found = False
for future in futures:
result = future.result()
main_window.update()
if 'PYPONGRST;SVRINFO' in result:
print(result)
raw_response = result.split(';')
response = []
for block in raw_response:
print(block)
chunks = block.split(':')
for chunk in chunks:
response.append(chunk)
print(response)
if 'NAME' in response:
session_name = response[response.index('NAME') + 1]
if not game_found:
listbox.delete(0, tk.END)
listbox.insert(0, session_name)
game_found = True
main_window.update()
if not game_found:
listbox.delete(0, tk.END)
listbox.insert(0, 'No games found!')
def get_addresses():
interface_info = client.get_interface_info()
shortened_info = []
for interface in interface_info:
shortened_info.append('{0} [{1}]'.format(interface.network, interface.interface_name))
return shortened_info
# Create the top title label
title_label = tk.Label(text='PynPong Launcher', fg='white', bg='#000040', padx=1, pady=20)
title_label.pack(side=tk.TOP, fill=tk.BOTH)
# Create the list of games
global listbox
listbox = tk.Listbox(main_window)
listbox.pack(side=tk.LEFT, expand=True, fill=tk.BOTH)
# Add scrollbar
@ -49,7 +110,13 @@ def main():
button_frame = tk.Frame(main_window)
tk.Button(button_frame, text='Host a Game', height=3, width=30).pack()
tk.Button(button_frame, text='Join Selected Game', height=3, width=30).pack()
tk.Button(button_frame, text='Refresh Game List', height=2, width=20, command=get_games).pack()
tk.Button(button_frame, text='Refresh Game List', height=2, width=20,
command=lambda: get_games(selected_network)).pack()
networks = get_addresses()
selected_network = tk.StringVar()
selected_network.set(networks[0])
interface_menu = tk.OptionMenu(button_frame, selected_network, *networks)
interface_menu.pack()
button_frame.pack(side=tk.RIGHT)
# Set it so that if the X is pressed the application quits