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, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QMainWindow, QPlainTextEdit, QPushButton, QTableWidget, QTableWidgetItem, QTabWidget, QVBoxLayout, QWidget, ) from fake_player import FakePlayerManager from server_core import FalloutTogetherServer from world_state_presets import TIME_PRESETS, WEATHER_PRESETS, format_hhmm_label 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.world_state_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") tabs.addTab(self._create_weather_time_tab(), "Weather / Time") 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_weather_time_tab(self) -> QWidget: tab = QWidget() layout = QVBoxLayout(tab) status_group = QGroupBox("Server World State") status_layout = QGridLayout(status_group) status_fields = [ ("time", "Current time (HHmm)"), ("weather", "Current weather"), ] for index, (key, label_text) in enumerate(status_fields): label = QLabel(f"{label_text}:") value = QLabel("Not set") value.setWordWrap(True) self.world_state_labels[key] = value status_layout.addWidget(label, index, 0) status_layout.addWidget(value, index, 1) layout.addWidget(status_group) weather_group = QGroupBox("Set Weather") weather_layout = QGridLayout(weather_group) for index, (label, fw_console_arg) in enumerate(WEATHER_PRESETS): button = QPushButton(label) button.setToolTip(f"Runs `fw {fw_console_arg}` on the current world-state host.") button.clicked.connect( lambda _checked=False, console_arg=fw_console_arg: self._set_server_weather(console_arg) ) weather_layout.addWidget(button, index // 3, index % 3) layout.addWidget(weather_group) time_group = QGroupBox("Set Time (set gamehour to HHmm)") time_layout = QGridLayout(time_group) for index, (label, hhmm) in enumerate(TIME_PRESETS): display = f"{label} ({format_hhmm_label(hhmm)})" button = QPushButton(display) button.setToolTip(f"Runs `set gamehour to {hhmm}` on all connected clients.") button.clicked.connect(lambda _checked=False, value=hhmm: self._set_server_time(value)) time_layout.addWidget(button, index // 3, index % 3) layout.addWidget(time_group) hint = QLabel( "Time is sent to all clients. Weather runs `fw <8-digit-id>` on the current world-state host only, " "then the host relays weather to other clients. Edit preset IDs in " "server/world_state_presets.py to match what works in your console." ) hint.setWordWrap(True) layout.addWidget(hint) layout.addStretch() return tab def _set_server_weather(self, fw_console_arg: str) -> None: if not self.server.is_running(): self._append_log("Start the server before setting weather.") return if self.server.set_server_weather(fw_console_arg): self._append_log(f"Requested server weather command: fw {fw_console_arg}") self._refresh_server_state() def _set_server_time(self, hhmm: str) -> None: if not self.server.is_running(): self._append_log("Start the server before setting time.") return if self.server.set_server_time(hhmm): self._append_log(f"Requested server time: {hhmm}") self._refresh_server_state() 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(self._format_host_stat(stats)) 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() self._refresh_world_state_labels() def _refresh_world_state_labels(self) -> None: world_state = self.server.get_server_world_state() time_label = self.world_state_labels.get("time") weather_label = self.world_state_labels.get("weather") if time_label is not None: hhmm = world_state.get("timeHHmm") if hhmm: time_label.setText(f"{hhmm} ({format_hhmm_label(hhmm)})") else: time_label.setText("Not set") if weather_label is not None: console_arg = world_state.get("weatherConsoleArg") form_id = world_state.get("weatherFormId") if console_arg: suffix = f" (relay 0x{form_id})" if form_id else "" weather_label.setText(f"fw {console_arg}{suffix}") else: weather_label.setText("Not set") 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_host_stat(stats: dict[str, Any]) -> str: host = str(stats.get("host", "-")) port = stats.get("port") lan_addresses = stats.get("lanAddresses") or [] if not lan_addresses or port is None: return host connect_targets = ", ".join(f"{address}:{port}" for address in lan_addresses) return f"{host} (LAN: {connect_targets})" @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())