pygame client/server example with sockets

client.py

import pygame as py
import socket as sk
import threading as th


class Client:
    __slots__ = (
        "__host",
        "__port",
        "__s",
        "__running",
        "__connected",
        "__th_receive_data",
        "__th_pygame_loop",
    )

    def __init__(self, **kwargs):
        self.__host = kwargs.get("host", "localhost")
        self.__port = kwargs.get("port", 8080)
        self.__s = sk.socket(sk.AF_INET, sk.SOCK_STREAM)
        self.__running = False
        self.__connected = False
        self.__th_receive_data = th.Thread(target=self.__receive_data)
        self.__th_pygame_loop = th.Thread(target=self.__pygame_loop)

    def connect(self) -> None:
        """Connect to the server."""
        try:
            self.__s.connect((self.__host, self.__port))
        except sk.error as exc:
            raise RuntimeError(
                f"Could not connect to {self.__host}:{self.__port}"
            ) from exc

    def __receive_data(self) -> None:
        """Receive data from the server."""
        print("Waiting for data...")
        while self.__connected:
            data = self.__s.recv(1024)
            print(f"Received data: {data}")
            if data == b"exit_server":
                print("Server closed connection.")
                self.close_connection()

    def __send_data(self, data: bytes) -> None:
        """Send data to the server."""
        self.__s.send(data)

    def close_connection(self) -> None:
        """Close the connection to the server."""
        self.__connected = False
        self.__s.close()
        print("Connection closed!")

    def close_pygame(self) -> None:
        """Close the pygame window."""
        self.__running = False
        py.quit()
        print("Closing GUI...")

    def __pygame_loop(self) -> None:
        """Pygame loop."""
        py.init()
        screen = py.display.set_mode((400, 300))
        self.__running = True
        while self.__running:
            screen.fill((255, 255, 255))
            py.display.flip()
            for event in py.event.get():
                if event.type == py.QUIT:
                    if self.__connected:
                        self.__send_data(b"exit")
                    self.close_pygame()
        print("Pygame closed!")

    def start(self) -> None:
        """Start the client."""
        self.__connected = True
        self.__th_receive_data.start()
        self.__th_pygame_loop.start()
        self.__send_data(b"Hello from client!")
        print("Client started!")


if __name__ == "__main__":
    client = Client()
    client.connect()
    client.start()

server.py

import socket as sk
import threading as th
import typing as t


class Server:
    __slots__ = (
        "__host",
        "__port",
        "__max_connections",
        "__s",
        "__running",
        "__clients",
        "__th_wait_clients",
    )

    def __init__(self, **kwargs):
        self.__host = kwargs.get("host", "localhost")
        self.__port = kwargs.get("port", 8080)
        self.__max_connections = kwargs.get("max_connections", 5)
        self.__s = sk.socket(sk.AF_INET, sk.SOCK_STREAM)
        self.__running = False
        self.__clients = list()
        self.__th_wait_clients = th.Thread(target=self.__wait_clients)

    @property
    def address(self) -> t.Tuple[str, int]:
        return self.__host, self.__port

    def __initialize(self) -> None:
        """Initialize the server."""
        try:
            self.__s.bind(self.address)
            self.__s.listen(self.__max_connections)
        except sk.error as exc:
            raise RuntimeError(
                f"Could not listen on {self.__host}:{self.__port}"
            ) from exc

    def __wait_clients(self) -> None:
        """Wait for clients to connect."""
        print("Waiting for clients...")
        while self.__running:
            try:
                client, address = self.__s.accept()
                self.__clients.append(client)
                print(f"New connection from: {address}")
                client_handler = th.Thread(target=self.__handle_client, args=(client,))
                client_handler.start()
            except sk.error as exc:
                print(f"An error occurred: {exc}")

    def __handle_client(self, client: sk.socket) -> None:
        client.send(b"Welcome to Pygame server!")
        # clients first message, should be the username, id or something else to identify the client.
        data = client.recv(1024)
        while True:
            try:
                if not data:
                    break
                if data == b"exit":
                    self.__remove_client(client)
                    break
                data = client.recv(1024)
                print(f"Received data: {data} from {client.getpeername()}")
            except sk.error as exc:
                print(f"An error occurred: {exc}")
                self.__remove_client(client)
                break
        if client in self.__clients:
            self.__remove_client(client)

    def __remove_client(self, client: sk.socket) -> None:
        """Removes a client from the list of clients."""
        self.__clients.remove(client)
        client.send(b"exit_server")  # send exit signal to client
        print(f"Client {client.getpeername()} disconnected")

    def __send_message(self) -> None:
        """Send a message to all clients."""
        message = input()
        if message == "exit":
            self.stop()
            return
        elif message == "":
            return
        elif message == "list":  # example command for the server.
            for client in self.__clients:
                print(f"Client {client.getpeername()}")
            return
        for client in self.__clients:
            print(f"Sending message to {client.getpeername()}")
            client.send(message.encode("utf-8"))

    def stop(self) -> None:
        """Stop the server."""
        self.__running = False
        print("Server stopped")
        self.__s.close()

    def run(self) -> None:
        """Run the server."""
        if self.__running:
            raise RuntimeError("Server is already running")
        self.__initialize()
        self.__running = True
        self.__th_wait_clients.start()
        while self.__running:
            try:
                self.__send_message()
            except KeyboardInterrupt:
                self.stop()
            except Exception as exc:
                self.stop()
                raise RuntimeError("An error occurred") from exc


if __name__ == "__main__":
    server = Server()
    server.run()