Edit on GitHub

mousefox.examples.chat

Example Chat room for MouseFox.

View Source

  1"""Example Chat room for MouseFox."""
  2
  3import arrow
  4import json
  5import pgnet
  6from dataclasses import dataclass, field
  7from pgnet import Packet, Response
  8import kvex as kx
  9
 10
 11BLANK_DATA = dict(message_log=list())
 12
 13
 14@dataclass
 15class Message:
 16    """Chat message."""
 17    username: str
 18    text: str
 19    time: float = field(default_factory=lambda: arrow.now().timestamp())
 20
 21    def serialize(self) -> str:
 22        """Export to string."""
 23        data = dict(username=self.username, text=self.text, time=self.time)
 24        return json.dumps(data)
 25
 26    @classmethod
 27    def deserialize(cls, raw_data: str, /) -> "Message":
 28        """Import from string."""
 29        return cls(**json.loads(raw_data))
 30
 31
 32class Game(pgnet.Game):
 33    """Chat room logic."""
 34
 35    def __init__(self, name: str, *args, **kwargs):
 36        """Override base method."""
 37        self.name = name
 38        initial_message = Message("admin", f"Welcome to {name!r} chat room")
 39        self.message_log: list[Message] = [initial_message]
 40        self.users: set[str] = set()
 41
 42    @property
 43    def persistent(self) -> bool:
 44        """Override base property."""
 45        if len(self.message_log) <= 1:
 46            return False
 47        time_since_last_message = arrow.now().timestamp() - self.message_log[-1].time
 48        expired = time_since_last_message > 7200  # Over 2 hours
 49        return not expired
 50
 51    def get_lobby_info(self) -> str:
 52        """Override base method."""
 53        mtime = arrow.get(self.message_log[-1].time).to("utc").format("HH:mm:ss")
 54        return f"Last message: {mtime} (UTC)"
 55
 56    def user_joined(self, username: str):
 57        """Override base method."""
 58        self.users.add(username)
 59
 60    def user_left(self, username: str):
 61        """Override base method."""
 62        if username in self.users:
 63            self.users.remove(username)
 64
 65    def handle_game_packet(self, packet: Packet) -> Response:
 66        """Override base method."""
 67        text = packet.payload.get("text")
 68        if not text:
 69            return Response("Expected text in payload.", status=pgnet.Status.UNEXPECTED)
 70        message = Message(username=packet.username, text=text)
 71        self.message_log.append(message)
 72        return Response("Added message.")
 73
 74    def handle_heartbeat(self, packet: Packet) -> Response:
 75        """Override base method."""
 76        update_hash = self._update_hash
 77        client_hash = packet.payload.get("update_hash", -1)
 78        if client_hash == update_hash:
 79            return Response("Up to date.", dict(update_hash=update_hash))
 80        payload = dict(
 81            update_hash=update_hash,
 82            room_name=self.name,
 83            users=list(self.users),
 84            messages=[m.serialize() for m in self.message_log[-50:]],
 85        )
 86        return Response("Last 50 messages.", payload)
 87
 88    @property
 89    def _update_hash(self) -> int:
 90        data = (
 91            self.name,
 92            str(sorted(self.users)),
 93            self.message_log[-1].time,
 94        )
 95        return hash(data)
 96
 97
 98AWAITING_DATA_TEXT = "Awaiting data from server..."
 99
100
101class GameWidget(kx.XFrame):
102    """Chat GUI widget."""
103
104    def __init__(self, client: pgnet.Client, **kwargs):
105        """Override base method."""
106        super().__init__(**kwargs)
107        self.client = client
108        self._game_data = dict(
109            update_hash=None,
110            room_name=AWAITING_DATA_TEXT,
111            users=[],
112            messages=[Message("system", AWAITING_DATA_TEXT).serialize()],
113        )
114        self._make_widgets()
115        client.on_heartbeat = self.on_heartbeat
116        client.heartbeat_payload = self.heartbeat_payload
117
118    def on_subtheme(self, *args, **kwargs):
119        """Refresh widgets."""
120        super().on_subtheme(*args, **kwargs)
121        self._refresh_widgets()
122
123    def heartbeat_payload(self) -> dict:
124        """Override base method."""
125        return dict(update_hash=self._game_data["update_hash"])
126
127    def on_heartbeat(self, heartbeat_response: pgnet.Response):
128        """Override base method."""
129        server_hash = heartbeat_response.payload.get("update_hash")
130        our_hash = self._game_data["update_hash"]
131        if not server_hash or our_hash == server_hash:
132            return
133        self._game_data = heartbeat_response.payload
134        self._refresh_widgets()
135
136    def _refresh_widgets(self, *args):
137        room_name = self._game_data["room_name"]
138        users = set(self._game_data["users"])
139        info_fg2 = self.app.theme.secondary.fg2.markup
140        bullet = self.app.theme.secondary.accent2.markup("•")
141        self.info_panel.text = "\n".join([
142            f"[u][b]Chat Room[/b][/u]\n[i]{info_fg2(room_name)}[/i]",
143            "\n",
144            "[u][b]Users[/b][/u]",
145            *(f" {bullet} {info_fg2(user)}" for user in users),
146        ])
147        text_lines = []
148        chevron = self.subtheme.accent1.markup(">>>")
149        for raw_message in self._game_data["messages"]:
150            message = Message.deserialize(raw_message)
151            time = arrow.get(message.time).to("local").format("HH:mm:ss")
152            is_author = message.username == self.client._username
153            color = kx.XColor.from_hex("77ff77" if is_author else "ff7777")
154            text_lines.append(color.markup(f"[u]{time} | {message.username}[/u]"))
155            text_lines.append(f"{chevron} {message.text}")
156        self.messages_label.text = "\n".join(text_lines)
157
158    def _make_widgets(self):
159        with self.app.subtheme_context("secondary"):
160            self.info_panel = kx.XLabel(
161                text="Getting chat room info...",
162                halign="left",
163                valign="top",
164            )
165            info_frame = kx.pwrap(kx.fwrap(kx.pwrap(self.info_panel)))
166            info_frame.set_size(hx=0.3)
167        self.messages_label = kx.XLabel(
168            text="Getting chat messages...",
169            halign="left",
170            valign="bottom",
171            fixed_width=True,
172        )
173        with self.app.subtheme_context("accent"):
174            self.message_input = kx.XInput(on_text_validate=self._message_validate)
175            self.message_input.focus = True
176            input_frame = kx.pwrap(kx.fwrap(self.message_input))
177            input_frame.set_size(y="55dp")
178        messages_frame = kx.pwrap(kx.XScroll(self.messages_label))
179        chat_frame = kx.XBox(orientation="vertical")
180        chat_frame.add_widgets(messages_frame, input_frame)
181        main_frame = kx.XBox()
182        main_frame.add_widgets(info_frame, kx.pwrap(chat_frame))
183        self.clear_widgets()
184        self.add_widget(main_frame)
185
186    def _message_validate(self, w):
187        self.client.send(pgnet.Packet("message", dict(text=w.text)))
188        w.text = ""
189
190
191INFO_TEXT = (
192    "[b][u]Welcome to MouseFox[/u][/b]"
193    "\n\n"
194    "This chat server is a builtin example to demo MouseFox."
195)
196ONLINE_INFO_TEXT = (
197    "[u]Connecting to a server[/u]"
198    "\n\n"
199    "To register (if the server allows it) simply choose a username and password"
200    " and log in."
201)
202APP_CONFIG = dict(
203    game_class=Game,
204    game_widget=GameWidget,
205    title="MouseFox chat",
206    info_text=INFO_TEXT,
207    online_info_text=ONLINE_INFO_TEXT,
208)
209
210
211def run():
212    """Run chat example."""
213    from .. import run
214
215    run(**APP_CONFIG)

@dataclass

class Message: View Source

15@dataclass
16class Message:
17    """Chat message."""
18    username: str
19    text: str
20    time: float = field(default_factory=lambda: arrow.now().timestamp())
21
22    def serialize(self) -> str:
23        """Export to string."""
24        data = dict(username=self.username, text=self.text, time=self.time)
25        return json.dumps(data)
26
27    @classmethod
28    def deserialize(cls, raw_data: str, /) -> "Message":
29        """Import from string."""
30        return cls(**json.loads(raw_data))

Chat message.

Message(username: str, text: str, time: float = <factory>)

def serialize(self) -> str: View Source

22    def serialize(self) -> str:
23        """Export to string."""
24        data = dict(username=self.username, text=self.text, time=self.time)
25        return json.dumps(data)

Export to string.

@classmethod

def deserialize(cls, raw_data: str, /) -> mousefox.examples.chat.Message: View Source

27    @classmethod
28    def deserialize(cls, raw_data: str, /) -> "Message":
29        """Import from string."""
30        return cls(**json.loads(raw_data))

Import from string.

class Game(pgnet.util.Game): View Source

33class Game(pgnet.Game):
34    """Chat room logic."""
35
36    def __init__(self, name: str, *args, **kwargs):
37        """Override base method."""
38        self.name = name
39        initial_message = Message("admin", f"Welcome to {name!r} chat room")
40        self.message_log: list[Message] = [initial_message]
41        self.users: set[str] = set()
42
43    @property
44    def persistent(self) -> bool:
45        """Override base property."""
46        if len(self.message_log) <= 1:
47            return False
48        time_since_last_message = arrow.now().timestamp() - self.message_log[-1].time
49        expired = time_since_last_message > 7200  # Over 2 hours
50        return not expired
51
52    def get_lobby_info(self) -> str:
53        """Override base method."""
54        mtime = arrow.get(self.message_log[-1].time).to("utc").format("HH:mm:ss")
55        return f"Last message: {mtime} (UTC)"
56
57    def user_joined(self, username: str):
58        """Override base method."""
59        self.users.add(username)
60
61    def user_left(self, username: str):
62        """Override base method."""
63        if username in self.users:
64            self.users.remove(username)
65
66    def handle_game_packet(self, packet: Packet) -> Response:
67        """Override base method."""
68        text = packet.payload.get("text")
69        if not text:
70            return Response("Expected text in payload.", status=pgnet.Status.UNEXPECTED)
71        message = Message(username=packet.username, text=text)
72        self.message_log.append(message)
73        return Response("Added message.")
74
75    def handle_heartbeat(self, packet: Packet) -> Response:
76        """Override base method."""
77        update_hash = self._update_hash
78        client_hash = packet.payload.get("update_hash", -1)
79        if client_hash == update_hash:
80            return Response("Up to date.", dict(update_hash=update_hash))
81        payload = dict(
82            update_hash=update_hash,
83            room_name=self.name,
84            users=list(self.users),
85            messages=[m.serialize() for m in self.message_log[-50:]],
86        )
87        return Response("Last 50 messages.", payload)
88
89    @property
90    def _update_hash(self) -> int:
91        data = (
92            self.name,
93            str(sorted(self.users)),
94            self.message_log[-1].time,
95        )
96        return hash(data)

Chat room logic.

Game(name: str, *args, **kwargs) View Source

36    def __init__(self, name: str, *args, **kwargs):
37        """Override base method."""
38        self.name = name
39        initial_message = Message("admin", f"Welcome to {name!r} chat room")
40        self.message_log: list[Message] = [initial_message]
41        self.users: set[str] = set()

Override base method.

persistent: bool

Override base property.

def get_lobby_info(self) -> str: View Source

52    def get_lobby_info(self) -> str:
53        """Override base method."""
54        mtime = arrow.get(self.message_log[-1].time).to("utc").format("HH:mm:ss")
55        return f"Last message: {mtime} (UTC)"

Override base method.

def user_joined(self, username: str): View Source

57    def user_joined(self, username: str):
58        """Override base method."""
59        self.users.add(username)

Override base method.

def user_left(self, username: str): View Source

61    def user_left(self, username: str):
62        """Override base method."""
63        if username in self.users:
64            self.users.remove(username)

Override base method.

def handle_game_packet(self, packet: pgnet.util.Packet) -> pgnet.util.Response: View Source

66    def handle_game_packet(self, packet: Packet) -> Response:
67        """Override base method."""
68        text = packet.payload.get("text")
69        if not text:
70            return Response("Expected text in payload.", status=pgnet.Status.UNEXPECTED)
71        message = Message(username=packet.username, text=text)
72        self.message_log.append(message)
73        return Response("Added message.")

Override base method.

def handle_heartbeat(self, packet: pgnet.util.Packet) -> pgnet.util.Response: View Source

75    def handle_heartbeat(self, packet: Packet) -> Response:
76        """Override base method."""
77        update_hash = self._update_hash
78        client_hash = packet.payload.get("update_hash", -1)
79        if client_hash == update_hash:
80            return Response("Up to date.", dict(update_hash=update_hash))
81        payload = dict(
82            update_hash=update_hash,
83            room_name=self.name,
84            users=list(self.users),
85            messages=[m.serialize() for m in self.message_log[-50:]],
86        )
87        return Response("Last 50 messages.", payload)

Override base method.

Inherited Members

pgnet.util.Game: heartbeat_rate; handle_packet; get_save_string; update

class GameWidget(kvex.widgets.layouts.XFrame): View Source

102class GameWidget(kx.XFrame):
103    """Chat GUI widget."""
104
105    def __init__(self, client: pgnet.Client, **kwargs):
106        """Override base method."""
107        super().__init__(**kwargs)
108        self.client = client
109        self._game_data = dict(
110            update_hash=None,
111            room_name=AWAITING_DATA_TEXT,
112            users=[],
113            messages=[Message("system", AWAITING_DATA_TEXT).serialize()],
114        )
115        self._make_widgets()
116        client.on_heartbeat = self.on_heartbeat
117        client.heartbeat_payload = self.heartbeat_payload
118
119    def on_subtheme(self, *args, **kwargs):
120        """Refresh widgets."""
121        super().on_subtheme(*args, **kwargs)
122        self._refresh_widgets()
123
124    def heartbeat_payload(self) -> dict:
125        """Override base method."""
126        return dict(update_hash=self._game_data["update_hash"])
127
128    def on_heartbeat(self, heartbeat_response: pgnet.Response):
129        """Override base method."""
130        server_hash = heartbeat_response.payload.get("update_hash")
131        our_hash = self._game_data["update_hash"]
132        if not server_hash or our_hash == server_hash:
133            return
134        self._game_data = heartbeat_response.payload
135        self._refresh_widgets()
136
137    def _refresh_widgets(self, *args):
138        room_name = self._game_data["room_name"]
139        users = set(self._game_data["users"])
140        info_fg2 = self.app.theme.secondary.fg2.markup
141        bullet = self.app.theme.secondary.accent2.markup("•")
142        self.info_panel.text = "\n".join([
143            f"[u][b]Chat Room[/b][/u]\n[i]{info_fg2(room_name)}[/i]",
144            "\n",
145            "[u][b]Users[/b][/u]",
146            *(f" {bullet} {info_fg2(user)}" for user in users),
147        ])
148        text_lines = []
149        chevron = self.subtheme.accent1.markup(">>>")
150        for raw_message in self._game_data["messages"]:
151            message = Message.deserialize(raw_message)
152            time = arrow.get(message.time).to("local").format("HH:mm:ss")
153            is_author = message.username == self.client._username
154            color = kx.XColor.from_hex("77ff77" if is_author else "ff7777")
155            text_lines.append(color.markup(f"[u]{time} | {message.username}[/u]"))
156            text_lines.append(f"{chevron} {message.text}")
157        self.messages_label.text = "\n".join(text_lines)
158
159    def _make_widgets(self):
160        with self.app.subtheme_context("secondary"):
161            self.info_panel = kx.XLabel(
162                text="Getting chat room info...",
163                halign="left",
164                valign="top",
165            )
166            info_frame = kx.pwrap(kx.fwrap(kx.pwrap(self.info_panel)))
167            info_frame.set_size(hx=0.3)
168        self.messages_label = kx.XLabel(
169            text="Getting chat messages...",
170            halign="left",
171            valign="bottom",
172            fixed_width=True,
173        )
174        with self.app.subtheme_context("accent"):
175            self.message_input = kx.XInput(on_text_validate=self._message_validate)
176            self.message_input.focus = True
177            input_frame = kx.pwrap(kx.fwrap(self.message_input))
178            input_frame.set_size(y="55dp")
179        messages_frame = kx.pwrap(kx.XScroll(self.messages_label))
180        chat_frame = kx.XBox(orientation="vertical")
181        chat_frame.add_widgets(messages_frame, input_frame)
182        main_frame = kx.XBox()
183        main_frame.add_widgets(info_frame, kx.pwrap(chat_frame))
184        self.clear_widgets()
185        self.add_widget(main_frame)
186
187    def _message_validate(self, w):
188        self.client.send(pgnet.Packet("message", dict(text=w.text)))
189        w.text = ""

Chat GUI widget.

GameWidget(client: pgnet.client.Client, **kwargs) View Source

105    def __init__(self, client: pgnet.Client, **kwargs):
106        """Override base method."""
107        super().__init__(**kwargs)
108        self.client = client
109        self._game_data = dict(
110            update_hash=None,
111            room_name=AWAITING_DATA_TEXT,
112            users=[],
113            messages=[Message("system", AWAITING_DATA_TEXT).serialize()],
114        )
115        self._make_widgets()
116        client.on_heartbeat = self.on_heartbeat
117        client.heartbeat_payload = self.heartbeat_payload

Override base method.

def on_subtheme(self, *args, **kwargs): View Source

119    def on_subtheme(self, *args, **kwargs):
120        """Refresh widgets."""
121        super().on_subtheme(*args, **kwargs)
122        self._refresh_widgets()

Refresh widgets.

def heartbeat_payload(self) -> dict: View Source

124    def heartbeat_payload(self) -> dict:
125        """Override base method."""
126        return dict(update_hash=self._game_data["update_hash"])

Override base method.

def on_heartbeat(self, heartbeat_response: pgnet.util.Response): View Source

128    def on_heartbeat(self, heartbeat_response: pgnet.Response):
129        """Override base method."""
130        server_hash = heartbeat_response.payload.get("update_hash")
131        our_hash = self._game_data["update_hash"]
132        if not server_hash or our_hash == server_hash:
133            return
134        self._game_data = heartbeat_response.payload
135        self._refresh_widgets()

Override base method.

Inherited Members

kvex.behaviors.XThemed: subtheme_name; subtheme; trigger_subtheme
kvex.widgets.layouts.XAnchor: wrap
kvex.widgets.widget.XWidget: add_widgets; set_size; set_focus; make_bg; app
kivy.uix.anchorlayout.AnchorLayout: padding; anchor_x; anchor_y; do_layout
kivy.uix.layout.Layout: add_widget; remove_widget; layout_hint_with_bounds
kivy.uix.widget.Widget: proxy_ref; apply_class_lang_rules; collide_point; collide_widget; on_motion; on_touch_down; on_touch_move; on_touch_up; on_kv_post; clear_widgets; register_for_motion_event; unregister_for_motion_event; export_to_png; export_as_image; get_root_window; get_parent_window; walk; walk_reverse; to_widget; to_window; to_parent; to_local; get_window_matrix; x; y; width; height; pos; size; get_right; set_right; right; get_top; set_top; top; get_center_x; set_center_x; center_x; get_center_y; set_center_y; center_y; center; cls; children; parent; size_hint_x; size_hint_y; size_hint; pos_hint; size_hint_min_x; size_hint_min_y; size_hint_min; size_hint_max_x; size_hint_max_y; size_hint_max; ids; opacity; on_opacity; canvas; get_disabled; set_disabled; inc_disabled; dec_disabled; disabled; motion_filter
kivy._event.EventDispatcher: register_event_type; unregister_event_types; unregister_event_type; is_event_type; bind; unbind; fbind; funbind; unbind_uid; get_property_observers; events; dispatch; dispatch_generic; dispatch_children; setter; getter; property; properties; create_property; apply_property

def run(): View Source

212def run():
213    """Run chat example."""
214    from .. import run
215
216    run(**APP_CONFIG)

Run chat example.