client.py

@@ -0,0 +1,136 @@
+# Kyler Olsen
+# CS 2450 Final Project
+# Apr 2025
+
+from __future__ import annotations
+import socket
+import select
+import network_utilities
+
+
+class Game:
+
+    __player: Player
+    __server: socket.socket
+
+    def __init__(self, player: Player, server: socket.socket):
+        self.__player = player
+        self.__server = server
+
+    def start_game(self):
+        data = network_utilities.pack_varint(2)
+        self.__server.send(data)
+
+    def start_round(self, difficulty: int):
+        data = network_utilities.pack_varint(3)
+        data += network_utilities.pack_varint(difficulty)
+        self.__server.send(data)
+
+    def guess_reference(self, url: str):
+        data = network_utilities.pack_varint(4)
+        data += network_utilities.pack_string(url)
+        self.__server.send(data)
+
+    def end_game(self):
+        data = network_utilities.pack_varint(5)
+        self.__server.send(data)
+
+    def update(self):
+        ready_to_read, _, _ = select.select([self.__server], [], [], 0)
+        if ready_to_read:
+            packet_id = network_utilities.unpack_varint(self.__server)
+            if packet_id == 1:
+                name = network_utilities.unpack_string(self.__server)
+                self.__player.player_joined(name)
+            elif packet_id == 2:
+                text = network_utilities.unpack_string(self.__server)
+                self.__player.new_verse(text)
+            elif packet_id == 3:
+                self.__player.guess_incorrect()
+            elif packet_id == 4:
+                self.__player.guess_correct()
+            elif packet_id == 5:
+                points = network_utilities.unpack_varint(self.__server)
+                url = network_utilities.unpack_string(self.__server)
+                player = network_utilities.unpack_string(self.__server)
+                self.__player.verse_guessed(points, url, player)
+            elif packet_id == 6:
+                players = network_utilities.unpack_string_array(self.__server)
+                scores = network_utilities.unpack_varint_array(self.__server)
+                self.__player.game_over(players, scores)
+
+
+class Player:
+
+    __name: str
+    __verse: str
+    __score: int
+    __game: Game | None
+
+    def __init__(self, name: str):
+        self.__name = name
+        self.__verse = ""
+        self.__score = 0
+        self.__game = None
+
+    @property
+    def name(self) -> str: return self.__name
+
+    @property
+    def verse(self) -> str: return self.__verse
+
+    @property
+    def score(self) -> int: return self.__score
+
+    def join_game(self, host: str = 'localhost', port: int = 7788):
+        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        conn.connect((host, port))
+        data = network_utilities.pack_varint(1)
+        data += network_utilities.pack_string(self.name)
+        conn.send(data)
+        self.__game = Game(self, conn)
+
+    def start_game(self):
+        if self.__game is not None:
+            self.__game.start_game()
+
+    def guess_reference(self, url: str):
+        if self.__game is not None:
+            self.__game.guess_reference(url)
+
+    def new_round(self, difficulty: int):
+        if self.__game is not None:
+            self.__game.start_round(difficulty)
+
+    def end_game(self):
+        if self.__game is not None:
+            self.__game.end_game()
+
+    def player_joined(self, name: str):
+        if self.__game is not None:
+            pass
+
+    def new_verse(self, text: str):
+        if self.__game is not None:
+            pass
+
+    def guess_incorrect(self):
+        if self.__game is not None:
+            pass
+
+    def guess_correct(self):
+        if self.__game is not None:
+            pass
+
+    def verse_guessed(self, points: int, url: str, player: str):
+        if self.__game is not None:
+            pass
+
+    def game_over(self, players: list[str], scores: list[int]):
+        if self.__game is not None:
+            pass
+
+    def update(self):
+        if self.__game is not None:
+            self.__game.update()
+

library.py

@@ -3,39 +3,67 @@
 # Apr 2025
 
 from __future__ import annotations
-from typing import TYPE_CHECKING
 import json
 import random
-
-if TYPE_CHECKING:
-    from game import Game
+import socket
+import select
+import network_utilities
+from server import Game
 
 
 class Library:
 
     __verses: dict
     __games: list[Game]
+    __host: str
+    __port: int
+    __socket: socket.socket
 
-    def __init__(self):
+    def __init__(self, host: str = '', port: int = 7788):
         with open("data/scripture-frequencies.json", encoding='utf-8') as file:
             self.__verses = json.load(file)
         self.__games = []
 
-    def join_game(self, name: str, game_num: int = -1):
-        if game_num == -1:
-            for i, game in enumerate(self.__games):
-                if not game.active:
-                    game_num = i
-                    break
-            else:
-                self.__games.append(Game())
-                game_num = len(self.__games) - 1
-        self.__games[game_num].add_player(name)
+        self.__host = host
+        self.__port = port
+        self.__socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
-    def get_verse(self, difficulty: int):
+    def serve_forever(self):
+        try:
+            with self.__socket as s:
+                s.bind((self.__host, self.__port))
+                s.listen(1)
+                s.setblocking(False)
+                while True:
+                    ready_to_read, _, _ = select.select([s], [], [], 0)
+                    for game in self.__games[:]:
+                        if not game.finished: game.update()
+                        else: self.__games.remove(game)
+                    if ready_to_read:
+                        conn, _ = s.accept()
+                        conn.setblocking(False)
+                        if network_utilities.unpack_varint(conn) == 1:
+                            name = network_utilities.unpack_string(conn)
+                            self.join_game(name, conn)
+        except KeyboardInterrupt:
+            print("KeyboardInterrupt\nExiting...")
+            return
+
+    def join_game(self, name: str, conn: socket.socket):
+        for i, game in enumerate(self.__games):
+            if not game.active:
+                game_num = i
+                break
+        else:
+            self.__games.append(Game(self))
+            game_num = len(self.__games) - 1
+        self.__games[game_num].add_player(name, conn)
+
+    def get_verse(self, difficulty: int, game: Game):
         url = self.__select_verse(difficulty)
         text = self.__get_verse_text(url)
-        print(text, end='\n\n')
+
+        game.new_verse(url, text)
 
     def __select_verse(self, difficulty: int) -> str:
 
@@ -47,11 +75,16 @@ class Library:
         lang = 'eng'
         volume, book_url, chapter_url, verse_url = url[1:].split('/')
 
-        if volume == 'ot': filename = f"data/{lang}.old-testament.json"
-        elif volume == 'nt': filename = f"data/{lang}.new-testament.json"
-        elif volume == 'bofm': filename = f"data/{lang}.book-of-mormon.json"
-        elif volume == 'dc-testament': filename = f"data/{lang}.doctrine-and-covenants.json"
-        elif volume == 'pgp': filename = f"data/{lang}.pearl-of-great-price.json"
+        if volume == 'ot':
+            filename = f"data/{lang}.old-testament.json"
+        elif volume == 'nt':
+            filename = f"data/{lang}.new-testament.json"
+        elif volume == 'bofm':
+            filename = f"data/{lang}.book-of-mormon.json"
+        elif volume == 'dc-testament':
+            filename = f"data/{lang}.doctrine-and-covenants.json"
+        elif volume == 'pgp':
+            filename = f"data/{lang}.pearl-of-great-price.json"
 
         with open(filename, encoding='utf-8') as file:
             data = json.load(file)
@@ -76,8 +109,3 @@ class Library:
 
         return difficulty_verses
 
-
-if __name__ == '__main__':
-
-    lib = Library()
-    for i in range(11): lib.get_verse(i)

network_utilities.py

@@ -0,0 +1,69 @@
+# Kyler Olsen
+# CS 2450 Final Project
+# Apr 2025
+
+from __future__ import annotations
+from typing import TYPE_CHECKING
+import struct
+
+if TYPE_CHECKING:
+    from socket import socket
+
+def pack_varint(data: int) -> bytes:
+    ordinal = b''
+    while True:
+        byte = data & 0x7F
+        data >>= 7
+        ordinal += struct.pack('B', byte | (0x80 if data > 0 else 0))
+        if data == 0:
+            break
+    return ordinal
+
+def unpack_varint(conn: socket) -> int:
+    data = 0
+    for i in range(5):
+        ordinal = conn.recv(1)
+        if len(ordinal) == 0:
+            break
+        byte = ord(ordinal)
+        data |= (byte & 0x7F) << 7*i
+        if not byte & 0x80:
+            break
+    return data
+
+def pack_string(text: str) -> bytes:
+    data = pack_varint(len(text))
+    data += text.encode('utf-8')
+    return data
+
+def unpack_string(conn: socket) -> str:
+    length = unpack_varint(conn)
+    data = conn.recv(length)
+    text = data.decode('utf-8')
+    return text
+
+def pack_varint_array(array: list[int]) -> bytes:
+    data = pack_varint(len(array))
+    for i in array:
+        data += pack_varint(i)
+    return data
+
+def unpack_varint_array(conn: socket) -> list[int]:
+    length = unpack_varint(conn)
+    array: list[int] = []
+    for _ in range(length):
+        array.append(unpack_varint(conn))
+    return array
+
+def pack_string_array(array: list[str]) -> bytes:
+    data = pack_varint(len(array))
+    for i in array:
+        data += pack_string(i)
+    return data
+
+def unpack_string_array(conn: socket) -> list[str]:
+    length = unpack_varint(conn)
+    array: list[str] = []
+    for _ in range(length):
+        array.append(unpack_string(conn))
+    return array

server.py

@@ -0,0 +1,153 @@
+# Kyler Olsen
+# CS 2450 Final Project
+# Apr 2025
+
+from __future__ import annotations
+from typing import TYPE_CHECKING
+import select
+import datetime
+import network_utilities
+
+if TYPE_CHECKING:
+    from socket import socket
+    from library import Library
+
+
+class Game:
+
+    __library: Library
+    __current_url: str
+    __clients: list[Player]
+    __round_points: list[int]
+    __total_scores: list[int]
+    __active: bool
+    __finished: bool
+    __created: datetime.datetime
+
+    def __init__(self, library: Library):
+        self.__library = library
+        self.__current_url = ""
+        self.__clients = []
+        self.__round_points = []
+        self.__total_scores = []
+        self.__active = False
+        self.__finished = False
+        self.__created = datetime.datetime.now()
+
+    @property
+    def active(self) -> bool: return self.__active
+
+    @property
+    def finished(self) -> bool: return self.__finished
+
+    def add_player(self, name: str, conn: socket):
+        if not self.__active:
+            self.__clients.append(Player(name, self, conn))
+            self.__total_scores.append(0)
+            for player in self.__clients:
+                player.player_joined(name)
+
+    def start_game(self):
+        self.__active = True
+
+    def start_round(self, difficulty: int):
+        if self.__active and not self.__finished:
+            self.__round_points = [0] * len(self.__clients)
+            self.__library.get_verse(difficulty, self)
+
+    def new_verse(self, url: str, text: str):
+        for player in self.__clients:
+            self.__current_url = url
+            player.new_verse(text)
+
+    def guess_reference(self, url: str, player: Player):
+        if self.__active and not self.__finished:
+            if url == self.__current_url:
+                player.guess_correct()
+                for i, points in enumerate(self.__round_points):
+                    self.__total_scores[i] += points
+                    self.__clients[i].verse_guessed(
+                        points, self.__current_url, player.name)
+            else:
+                player.guess_incorrect()
+
+    def end_game(self):
+        self.__finished = True
+        for player in self.__clients:
+            player.game_over(
+                [i.name for i in self.__clients], self.__total_scores)
+
+    def update(self):
+        if not self.__active and (
+            datetime.datetime.now() - self.__created >=
+            datetime.timedelta(minutes=10)
+        ) or (
+            datetime.datetime.now() - self.__created >=
+            datetime.timedelta(1)
+        ): self.__finished = True
+        if not self.__finished:
+            for player in self.__clients:
+                player.update()
+
+
+class Player:
+
+    __name: str
+    __game: Game
+    __client: socket
+
+    def __init__(self, name: str, game: Game, conn: socket):
+        self.__name = name
+        self.__game = game
+        self.__client = conn
+
+    @property
+    def name(self) -> str: return self.__name
+
+    def player_joined(self, name: str):
+        data = network_utilities.pack_varint(1)
+        data += network_utilities.pack_string(name)
+        self.__client.send(data)
+
+    def new_verse(self, text: str):
+        data = network_utilities.pack_varint(2)
+        data += network_utilities.pack_string(text)
+        self.__client.send(data)
+
+    def guess_incorrect(self):
+        data = network_utilities.pack_varint(3)
+        self.__client.send(data)
+
+    def guess_correct(self):
+        data = network_utilities.pack_varint(4)
+        self.__client.send(data)
+
+    def verse_guessed(self, points: int, url: str, player: str):
+        data = network_utilities.pack_varint(5)
+        data += network_utilities.pack_varint(points)
+        data += network_utilities.pack_string(url)
+        data += network_utilities.pack_string(player)
+        self.__client.send(data)
+
+    def game_over(self, players: list[str], scores: list[int]):
+        data = network_utilities.pack_varint(6)
+        data += network_utilities.pack_string_array(players)
+        data += network_utilities.pack_varint_array(scores)
+        self.__client.send(data)
+        self.__client.close()
+
+    def update(self):
+        ready_to_read, _, _ = select.select([self.__client], [], [], 0)
+        if ready_to_read:
+            packet_id = network_utilities.unpack_varint(self.__client)
+            if packet_id == 2:
+                self.__game.start_game()
+            elif packet_id == 3:
+                difficulty = network_utilities.unpack_varint(self.__client)
+                self.__game.start_round(difficulty)
+            elif packet_id == 4:
+                url = network_utilities.unpack_string(self.__client)
+                self.__game.guess_reference(url, self)
+            elif packet_id == 5:
+                self.__game.end_game()
+