from __future__ import annotations import sys import time from typing import Any from PySide6.QtCore import QObject, Qt, QTimer, Signal from PySide6.QtGui import QCloseEvent from PySide6.QtWidgets import ( QApplication, QAbstractItemView, QGridLayout, QHBoxLayout, QHeaderView, QLabel, QMainWindow, QPlainTextEdit, QPushButton, QTableWidget, QTableWidgetItem, QTabWidget, QVBoxLayout, QWidget, ) from fake_player import FakePlayerManager from server_core import FalloutTogetherServer REFRESH_INTERVAL_MS = 500 class ServerLogBridge(QObject): message_received = Signal(str) class DevServerWindow(QMainWindow): def __init__(self) -> None: super().__init__() self.server = FalloutTogetherServer() self.log_bridge = ServerLogBridge() self.log_bridge.message_received.connect(self._append_log) self.server.add_log_listener(self._handle_server_log) self.fake_player_manager = FakePlayerManager( host=self.server.host, port=self.server.port, log_callback=self._handle_fake_client_log, client_snapshot_provider=lambda: self.server.get_clients(), ) self.stat_labels: dict[str, QLabel] = {} self.selected_fake_client_key: int | None = None self.setWindowTitle("Commonwealth Online Dev Server") self.resize(1100, 760) self.setCentralWidget(self._create_tabs()) self.refresh_timer = QTimer(self) self.refresh_timer.timeout.connect(self._refresh_server_state) self.refresh_timer.start(REFRESH_INTERVAL_MS) self._refresh_server_state() def closeEvent(self, event: QCloseEvent) -> None: self.refresh_timer.stop() self.fake_player_manager.stop_all() self.server.remove_log_listener(self._handle_server_log) self.server.stop() event.accept() def _create_tabs(self) -> QTabWidget: tabs = QTabWidget() tabs.addTab(self._create_server_console_tab(), "Server Console") tabs.addTab(self._create_fake_clients_tab(), "Fake Clients") return tabs def _create_server_console_tab(self) -> QWidget: tab = QWidget() layout = QVBoxLayout(tab) layout.addLayout(self._create_stats_layout()) layout.addWidget(QLabel("Console")) self.console = QPlainTextEdit() self.console.setReadOnly(True) layout.addLayout(self._create_server_button_layout()) layout.addWidget(self.console, stretch=2) layout.addWidget(QLabel("Connected Clients")) self.clients_table = QTableWidget() self.clients_table.setColumnCount(9) self.clients_table.setHorizontalHeaderLabels( [ "Player ID", "Address", "Port", "Connected for / connected time", "Last packet time", "Packets received", "Packets broadcast", "Last cellId", "Last position", ] ) self.clients_table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) self.clients_table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) self.clients_table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) self.clients_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) layout.addWidget(self.clients_table, stretch=1) return tab def _create_stats_layout(self) -> QGridLayout: layout = QGridLayout() stats = [ ("status", "Status"), ("host", "Host"), ("port", "Port"), ("uptime", "Uptime"), ("connectedClients", "Connected clients"), ("clientsConnected", "Total clients connected"), ("clientsDisconnected", "Total clients disconnected"), ("packetsReceived", "Packets received"), ("packetsBroadcast", "Packets broadcast"), ] for index, (key, label_text) in enumerate(stats): row = index // 3 column = (index % 3) * 2 label = QLabel(f"{label_text}:") value = QLabel("-") self.stat_labels[key] = value layout.addWidget(label, row, column) layout.addWidget(value, row, column + 1) return layout def _create_server_button_layout(self) -> QHBoxLayout: layout = QHBoxLayout() self.start_button = QPushButton("Start Server") self.start_button.clicked.connect(self._start_server) layout.addWidget(self.start_button) self.stop_button = QPushButton("Stop Server") self.stop_button.clicked.connect(self._stop_server) layout.addWidget(self.stop_button) self.clear_console_button = QPushButton("Clear Console") self.clear_console_button.clicked.connect(self.console.clear) layout.addWidget(self.clear_console_button) self.copy_console_button = QPushButton("Copy Console") self.copy_console_button.clicked.connect(self._copy_console) layout.addWidget(self.copy_console_button) layout.addStretch() return layout def _create_fake_clients_tab(self) -> QWidget: tab = QWidget() layout = QVBoxLayout(tab) button_layout = QHBoxLayout() self.add_fake_client_button = QPushButton("+ Add Fake Client") self.add_fake_client_button.clicked.connect(self._add_fake_client) button_layout.addWidget(self.add_fake_client_button) self.remove_fake_client_button = QPushButton("- Remove Selected") self.remove_fake_client_button.clicked.connect(self._remove_selected_fake_client) button_layout.addWidget(self.remove_fake_client_button) button_layout.addStretch() layout.addLayout(button_layout) action_layout = QHBoxLayout() self.set_idle_button = QPushButton("Set Idle") self.set_idle_button.clicked.connect(self._set_selected_fake_client_idle) action_layout.addWidget(self.set_idle_button) self.walk_to_player_button = QPushButton("Walk To Player") self.walk_to_player_button.clicked.connect(self._set_selected_fake_client_walk_to_player) action_layout.addWidget(self.walk_to_player_button) self.walk_circle_button = QPushButton("Walk Circle") self.walk_circle_button.clicked.connect(self._set_selected_fake_client_walk_circle) action_layout.addWidget(self.walk_circle_button) action_layout.addStretch() layout.addLayout(action_layout) script_action_layout = QHBoxLayout() self.jump_once_button = QPushButton("Jump Once") self.jump_once_button.clicked.connect(self._trigger_selected_fake_client_jump_once) script_action_layout.addWidget(self.jump_once_button) self.toggle_sneak_button = QPushButton("Toggle Sneak") self.toggle_sneak_button.clicked.connect(self._toggle_selected_fake_client_sneak) script_action_layout.addWidget(self.toggle_sneak_button) self.toggle_crouch_button = QPushButton("Toggle Crouch") self.toggle_crouch_button.clicked.connect(self._toggle_selected_fake_client_crouch) script_action_layout.addWidget(self.toggle_crouch_button) self.leave_cell_button = QPushButton("Leave Cell") self.leave_cell_button.clicked.connect(self._trigger_selected_fake_client_leave_cell) script_action_layout.addWidget(self.leave_cell_button) self.return_to_cell_button = QPushButton("Return To Cell") self.return_to_cell_button.clicked.connect(self._trigger_selected_fake_client_return_to_cell) script_action_layout.addWidget(self.return_to_cell_button) self.teleport_test_button = QPushButton("Teleport Test") self.teleport_test_button.clicked.connect(self._trigger_selected_fake_client_teleport_test) script_action_layout.addWidget(self.teleport_test_button) script_action_layout.addStretch() layout.addLayout(script_action_layout) self.fake_clients_table = QTableWidget() self.fake_clients_table.setColumnCount(6) self.fake_clients_table.setHorizontalHeaderLabels( ["Name", "Player ID", "Status", "Script", "Cell ID", "Position"] ) self.fake_clients_table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) self.fake_clients_table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) self.fake_clients_table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) self.fake_clients_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) layout.addWidget(self.fake_clients_table) # TODO: Add future fake-client scripts: Walk Square, Follow Player, # Sprint Toggle, Weapon Drawn Toggle, Disconnect After Delay, and # Multi-fake-player choreography. return tab def _start_server(self) -> None: try: self.server.start() except OSError as error: self._append_log(f"Failed to start server on {self.server.host}:{self.server.port}: {error}") except RuntimeError as error: self._append_log(f"Failed to start server: {error}") self._refresh_server_state() def _stop_server(self) -> None: if self.server.is_running(): self.fake_player_manager.stop_all() self.server.stop() self._append_log("Server stopped.") self._refresh_server_state() def _copy_console(self) -> None: QApplication.clipboard().setText(self.console.toPlainText()) def _add_fake_client(self) -> None: if not self.server.is_running(): self._append_log("Start the server before adding fake clients.") return self.fake_player_manager.add_client() self._refresh_fake_clients_table() def _remove_selected_fake_client(self) -> None: client_key = self._get_selected_fake_client_key() if client_key is None: self._append_log("No fake client selected.") return self.fake_player_manager.remove_client(client_key) self._refresh_fake_clients_table() def _set_selected_fake_client_idle(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.set_client_idle(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _set_selected_fake_client_walk_to_player(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.set_client_walk_to_player(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _set_selected_fake_client_walk_circle(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.set_client_walk_circle(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _trigger_selected_fake_client_jump_once(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.trigger_client_jump_once(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _toggle_selected_fake_client_sneak(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.toggle_client_sneak(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _toggle_selected_fake_client_crouch(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.toggle_client_crouch(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _trigger_selected_fake_client_leave_cell(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.trigger_client_leave_cell(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _trigger_selected_fake_client_return_to_cell(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.trigger_client_return_to_cell(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _trigger_selected_fake_client_teleport_test(self) -> None: client_key = self._get_selected_fake_client_key_for_control() if client_key is None: return if not self.fake_player_manager.trigger_client_teleport_test(client_key): self._append_log("Select a fake client first.") return self._refresh_fake_clients_table() def _get_selected_fake_client_key_for_control(self) -> int | None: client_key = self._get_selected_fake_client_key() if client_key is None: snapshots = self.fake_player_manager.get_snapshots() if len(snapshots) == 1: client_key = self._get_optional_int(snapshots[0].get("clientKey")) if client_key is not None: self.selected_fake_client_key = client_key self._append_log("No fake client row selected; using the only fake client.") if client_key is None: self._append_log("Select a fake client first.") return None if not self.server.is_running(): self._append_log("Start the server before controlling fake clients.") return None return client_key def _get_selected_fake_client_key(self) -> int | None: selected_items = self.fake_clients_table.selectedItems() if not selected_items: return None row = selected_items[0].row() name_item = self.fake_clients_table.item(row, 0) if name_item is None: return None client_key = name_item.data(Qt.ItemDataRole.UserRole) try: selected_key = int(client_key) self.selected_fake_client_key = selected_key return selected_key except (TypeError, ValueError): return None def _handle_server_log(self, message: str) -> None: self.log_bridge.message_received.emit(message) def _handle_fake_client_log(self, message: str) -> None: self.log_bridge.message_received.emit(message) def _append_log(self, message: str) -> None: timestamp = time.strftime("%H:%M:%S", time.localtime()) self.console.appendPlainText(f"[{timestamp}] {message}") def _refresh_server_state(self) -> None: stats = self.server.get_stats() clients = self.server.get_clients() is_running = bool(stats.get("isRunning", False)) self.stat_labels["status"].setText("Running" if is_running else "Stopped") self.stat_labels["host"].setText(str(stats.get("host", "-"))) self.stat_labels["port"].setText(str(stats.get("port", "-"))) self.stat_labels["uptime"].setText(self._format_duration(float(stats.get("uptimeSeconds", 0.0)))) self.stat_labels["connectedClients"].setText(str(stats.get("connectedClients", 0))) self.stat_labels["clientsConnected"].setText(str(stats.get("clientsConnected", 0))) self.stat_labels["clientsDisconnected"].setText(str(stats.get("clientsDisconnected", 0))) self.stat_labels["packetsReceived"].setText(str(stats.get("packetsReceived", 0))) self.stat_labels["packetsBroadcast"].setText(str(stats.get("packetsBroadcast", 0))) self.start_button.setEnabled(not is_running) self.stop_button.setEnabled(is_running) self._refresh_clients_table(clients) self._refresh_fake_clients_table() def _refresh_clients_table(self, clients: list[dict[str, Any]]) -> None: self.clients_table.setRowCount(len(clients)) now = time.time() for row, client in enumerate(clients): connected_at = self._get_optional_float(client.get("connectedAt")) last_packet_at = self._get_optional_float(client.get("lastPacketAt")) last_transform = client.get("lastTransform") if not isinstance(last_transform, dict): last_transform = {} connected_text = "-" if connected_at is not None: connected_text = f"{self._format_duration(now - connected_at)} / {self._format_timestamp(connected_at)}" values = [ client.get("playerId", "-"), client.get("address", "-"), client.get("port", "-"), connected_text, self._format_timestamp(last_packet_at) if last_packet_at is not None else "-", client.get("packetsReceived", 0), client.get("packetsBroadcast", 0), last_transform.get("cellId", "-"), self._format_position(last_transform), ] for column, value in enumerate(values): self.clients_table.setItem(row, column, self._create_table_item(str(value))) def _refresh_fake_clients_table(self) -> None: selected_key = self._get_selected_fake_client_key() if selected_key is None: selected_key = self.selected_fake_client_key snapshots = self.fake_player_manager.get_snapshots() self.fake_clients_table.setRowCount(len(snapshots)) row_to_select: int | None = None for row, snapshot in enumerate(snapshots): client_key = self._get_optional_int(snapshot.get("clientKey")) player_id = snapshot.get("playerId") values = [ snapshot.get("name", "-"), player_id if player_id is not None else "-", snapshot.get("status", "-"), snapshot.get("script", "-"), snapshot.get("cellId", "-"), self._format_fake_position(snapshot.get("position")), ] for column, value in enumerate(values): item = self._create_table_item(str(value)) if column == 0: item.setData(Qt.ItemDataRole.UserRole, client_key) self.fake_clients_table.setItem(row, column, item) if client_key is not None and client_key == selected_key: row_to_select = row if row_to_select is not None: self.fake_clients_table.selectRow(row_to_select) self.selected_fake_client_key = selected_key @staticmethod def _create_table_item(value: str) -> QTableWidgetItem: return QTableWidgetItem(value) @staticmethod def _get_optional_int(value: Any) -> int | None: try: return int(value) except (TypeError, ValueError): return None @staticmethod def _get_optional_float(value: Any) -> float | None: try: return float(value) except (TypeError, ValueError): return None @staticmethod def _format_duration(seconds: float) -> str: total_seconds = max(0, int(seconds)) hours, remainder = divmod(total_seconds, 3600) minutes, secs = divmod(remainder, 60) if hours: return f"{hours:d}:{minutes:02d}:{secs:02d}" return f"{minutes:d}:{secs:02d}" @staticmethod def _format_timestamp(timestamp: float) -> str: return time.strftime("%H:%M:%S", time.localtime(timestamp)) @staticmethod def _format_position(transform: dict[str, Any]) -> str: try: x = float(transform["x"]) y = float(transform["y"]) z = float(transform["z"]) except (KeyError, TypeError, ValueError): return "-" return f"{x:.2f}, {y:.2f}, {z:.2f}" @staticmethod def _format_fake_position(position: Any) -> str: try: x, y, z = position return f"{float(x):.1f}, {float(y):.1f}, {float(z):.1f}" except (TypeError, ValueError): return "-" def main() -> int: app = QApplication(sys.argv) window = DevServerWindow() window.show() return app.exec() if __name__ == "__main__": raise SystemExit(main())