pgnet.examples

 1"""A simple example implementation of a game and client.
 2
 3These examples are best viewed from [source](
 4https://github.com/ArielHorwitz/pgnet/blob/master/pgnet/examples.py).
 5"""
 6
 7import functools
 8from typing import Optional
 9from .client import Client
10from .util import Game, Packet, Response
11
12
13class ExampleGame(Game):
14    """A subclass of `pgnet.Game` as a simple example.
15
16    Maintains a log of strings, that users can get and add to.
17    """
18
19    persistent = True
20
21    def __init__(self, *args, save_string: Optional[str] = None, **kwargs):
22        """On initialization, import log from *save_string*."""
23        super().__init__(*args, **kwargs)
24        self.log = list(save_string.splitlines()) if save_string else ["Game started"]
25
26    def user_joined(self, username: str):
27        """Add log message on user join."""
28        self.log.append(f"Joined: {username}")
29
30    def user_left(self, username: str):
31        """Add log message on user leave."""
32        self.log.append(f"Left: {username}")
33
34    def handle_game_packet(self, packet: Packet) -> Response:
35        """Add log message and return full log."""
36        self.log.append(f"{packet.username} says: {packet.message!r}")
37        return Response("Message added.")
38
39    def handle_heartbeat(self, packet: Packet) -> Response:
40        """Return latest log messages."""
41        return Response("Latest log entries.", payload=dict(log=self.log[-10:]))
42
43    def get_save_string(self) -> str:
44        """Export the log."""
45        return "\n".join(self.log)
46
47
48class ExampleClient(Client):
49    """Client subclass to implement heartbeat for `ExampleGame`."""
50
51    log = []
52    on_status = functools.partial(print, ">> STATUS:")
53    """Print client status feedback."""
54    on_game = functools.partial(print, ">> GAME:")
55    """Print client game changes."""
56
57    def on_connection(self, connected: bool):
58        """Print connection changes and auto create a game when connected."""
59        print(">> CONNECTION:", connected)
60        if connected:
61            self.create_game("test")  # Will auto join if created
62            # self.join_game("test")  # Use this in case game already exists
63
64    def on_heartbeat(self, heartbeat: Response):
65        """Save and print the log if changed."""
66        game_log = heartbeat.payload.get("log", ["Empty log"])
67        if self.log != game_log:
68            self.log = game_log
69            print("New log:")
70            print("\n".join(f"-- {_}" for _ in game_log))
71
72
73__all__ = (
74    "ExampleGame",
75    "ExampleClient",
76)