Add Walk To Player script and GUI controls

Add developer tooling to make GUI-managed fake clients run a Walk To Player script and provide UI controls.

Changes:
- server/fake_player.py: implement SCRIPT_WALK_TO_PLAYER, thread-safe script switching, pose/angle tracking, walk-to-player movement (straight-line, +150 X offset), target selection via a walk target provider, configurable send rates/speed/stop distance, and robust parsing of server client snapshots. Sends moving transforms ~10 Hz and falls back to idle keep-alives. Logs when no target is available.
- server/dev_server_app.py: add Set Idle and Walk To Player buttons, safe selected-client handling, and pass a client snapshot provider callback (server.get_clients()) to the fake player manager.
- server/README.md & docs/dev-log.md: document the new controls, behavior, testing notes, known limitations, and next steps.

Notes:
- Walk To Player picks the lowest non-fake playerId with a valid lastTransform and targets that player's last transform +150 X.
- Movement is simple straight-line stepping (no pathfinding/navmesh/obstacle avoidance).
- No protocol, plugin, server core, or existing fake_client networking behavior was changed.
- Quick static check: py_compile ran for relevant modules; manual GUI and in-game validation are still required.
This commit is contained in:
2026-06-02 15:19:32 +12:00
parent 9db37d3436
commit 0b73b1eff1
4 changed files with 417 additions and 31 deletions
+57
View File
@@ -1038,6 +1038,63 @@ Fake Clients tab or when the GUI server stops.
--- ---
## 2026-06-02 - Fake Client Walk To Player
### Summary
Added Fake Clients tab movement controls and a first useful fake-client movement
script. GUI-managed fake clients can now be set to Idle or Walk To Player while
remaining ordinary external TCP clients that send normal transform packets
through the Python server.
### Files Changed
- `server/fake_player.py`
- `server/dev_server_app.py`
- `server/README.md`
- `docs/dev-log.md`
### Details
- Added `Idle` and `Walk To Player` script modes to `FakePlayerClient`.
- Added thread-safe script switching, pose tracking, and target state for
GUI-managed fake clients.
- Added `Set Idle` and `Walk To Player` buttons to the Fake Clients tab.
- Added safe selected-client handling with readable logs when no fake client is
selected or the server is stopped.
- Walk To Player reads `FalloutTogetherServer.get_clients()` snapshots through a
manager callback, filters out GUI-managed fake clients, and targets the lowest
real `playerId` with a valid `lastTransform`.
- Walk To Player moves in a straight line toward the real player's latest
transform plus a +150 X offset, sends moving transforms at about 10 Hz, and
returns to low-rate idle keep-alive transforms after reaching the stop
distance.
- No plugin code, protocol fields, `server/server.py`, or `server/fake_client.py`
were changed.
### Testing
- Ran `python -m py_compile server.py server_core.py client_session.py
fake_client.py dev_server_app.py fake_player.py` from `server`; it succeeded.
- Still needs manual GUI validation with `python dev_server_app.py`.
- Still needs in-game Fallout 4/F4SE validation in `F4TTestCell01`.
### Known Issues
- Walk To Player uses simple straight-line movement only; it does not perform
pathfinding, navmesh checks, or obstacle avoidance.
- Additional fake-client scripts are still future work.
### Next Steps
- Manually validate the GUI buttons with no real player connected and with a
Fallout 4 client in `F4TTestCell01`.
- Add future dev/test scripts such as Walk Circle, Walk Square, Follow Player,
Jump Loop, Sneak Toggle, Teleport Test, Leave Cell / Return, and Disconnect
After Delay.
---
## Entry Template ## Entry Template
Use this format for future updates: Use this format for future updates:
+38 -7
View File
@@ -44,6 +44,21 @@ clients for local relay testing. Click `Start Server`, then click
server-assigned `playerId`, appears in both GUI client tables, and sends one server-assigned `playerId`, appears in both GUI client tables, and sends one
idle transform per second for `F4TTestCell01`. idle transform per second for `F4TTestCell01`.
Select a fake client row to use the movement controls:
- `Set Idle`: returns the selected fake client to idle mode. It keeps its
current position, sends non-moving transforms once per second, and reports
`movementSpeed=0.0`.
- `Walk To Player`: makes the selected fake client walk in a straight line
toward the lowest connected real client with a valid transform snapshot. The
target point is offset by +150 units on X so the fake client does not stand
inside the real player. While moving, it sends normal transform packets at
about 10 Hz with `isMoving=true`; when it reaches the stop distance, it
returns to low-rate idle keep-alive transforms.
These controls are developer test tooling only. They do not change the network
protocol, server relay behavior, or Fallout 4 plugin behavior.
If the server is not running, the plugin should log a warning and Fallout 4 If the server is not running, the plugin should log a warning and Fallout 4
should continue launching normally. should continue launching normally.
@@ -56,8 +71,8 @@ should continue launching normally.
- `dev_server_app.py`: PySide6 developer GUI that starts/stops the reusable - `dev_server_app.py`: PySide6 developer GUI that starts/stops the reusable
server core, displays logs, shows live stats, and lists connected clients. server core, displays logs, shows live stats, and lists connected clients.
- `fake_player.py`: GUI fake-player test tooling. It owns background synthetic - `fake_player.py`: GUI fake-player test tooling. It owns background synthetic
TCP clients that connect through the normal server socket and send idle TCP clients that connect through the normal server socket, send transform
transform packets. packets, and run simple dev movement scripts such as Idle and Walk To Player.
- `fake_client.py`: lightweight receiver for testing welcome, transform, and - `fake_client.py`: lightweight receiver for testing welcome, transform, and
disconnect packets without launching a second Fallout 4 instance. disconnect packets without launching a second Fallout 4 instance.
- `requirements.txt`: optional Python dependencies for the desktop developer - `requirements.txt`: optional Python dependencies for the desktop developer
@@ -108,7 +123,8 @@ python dev_server_app.py
Click `Start Server`, then run `python fake_client.py` or launch Fallout 4 Click `Start Server`, then run `python fake_client.py` or launch Fallout 4
through F4SE. You can also click `+ Add Fake Client` to create a GUI-managed through F4SE. You can also click `+ Add Fake Client` to create a GUI-managed
synthetic client that sends an idle transform once per second. Select a fake synthetic client that sends an idle transform once per second. Select a fake
client row and click `- Remove Selected` to disconnect it cleanly. client row and click `Set Idle` or `Walk To Player` to control its script. Click
`- Remove Selected` to disconnect it cleanly.
Expected behavior: Expected behavior:
@@ -125,6 +141,11 @@ Expected behavior:
- GUI fake clients connect as ordinary external TCP clients, receive a - GUI fake clients connect as ordinary external TCP clients, receive a
`welcome` packet, store their assigned `playerId`, and send idle transforms `welcome` packet, store their assigned `playerId`, and send idle transforms
for cell `0B000F99`. for cell `0B000F99`.
- GUI fake clients can be switched between `Idle` and `Walk To Player` scripts
from the Fake Clients tab.
- Walk To Player uses `FalloutTogetherServer.get_clients()` snapshots to find
the lowest non-fake client with a valid last transform, then walks gradually
toward that player with a simple straight-line movement test.
- When a client disconnects, the server removes that client and broadcasts a - When a client disconnects, the server removes that client and broadcasts a
`disconnect` packet with the departed `playerId` to the remaining clients. `disconnect` packet with the departed `playerId` to the remaining clients.
- `fake_client.py` removes disconnected players from its in-memory - `fake_client.py` removes disconnected players from its in-memory
@@ -164,11 +185,17 @@ python dev_server_app.py
Click `Start Server`, then connect `fake_client.py` or the Fallout 4 plugin. Click `Start Server`, then connect `fake_client.py` or the Fallout 4 plugin.
To validate GUI fake clients, click `+ Add Fake Client` and confirm the fake To validate GUI fake clients, click `+ Add Fake Client` and confirm the fake
client appears in the Fake Clients table and Connected Clients table with an client appears in the Fake Clients table and Connected Clients table with an
assigned player ID. Click `- Remove Selected` and confirm it disconnects. assigned player ID. Select the fake client, click `Set Idle`, and confirm the
Script column stays or changes to `Idle`. Click `Walk To Player` with no real
Fallout 4 client connected and confirm the GUI logs that no real player target
is available. Click `- Remove Selected` and confirm it disconnects.
For in-game validation, launch Fallout 4 through F4SE and confirm the plugin For in-game validation, launch Fallout 4 through F4SE and confirm the plugin
connects, receives a `welcome` packet, sends transforms, and receives relayed connects, receives a `welcome` packet, sends transforms, and receives relayed
remote transforms or disconnect packets when another client is connected. remote transforms or disconnect packets when another client is connected. In
`F4TTestCell01`, select a GUI fake client and click `Walk To Player`; the fake
client should walk smoothly toward the real player's latest transform offset and
then stop near that offset.
## Implemented ## Implemented
@@ -187,6 +214,9 @@ remote transforms or disconnect packets when another client is connected.
- PySide6 developer server GUI with live logs, stats, connected-client table, - PySide6 developer server GUI with live logs, stats, connected-client table,
and a Fake Clients tab that can spawn idle synthetic TCP clients and a Fake Clients tab that can spawn idle synthetic TCP clients
- GUI-managed fake clients that receive welcome packets and send idle transforms - GUI-managed fake clients that receive welcome packets and send idle transforms
- Fake Clients tab movement controls for `Set Idle` and `Walk To Player`
- GUI-managed fake clients that can walk gradually toward a real Fallout 4
client using normal transform packets
## Working In Fake Client Only ## Working In Fake Client Only
@@ -197,8 +227,9 @@ remote transforms or disconnect packets when another client is connected.
## Planned Later ## Planned Later
- Fake client management - Additional fake client scripts such as Walk Circle, Walk Square, Follow
- Fake client movement scripts Player, Jump Loop, Sneak Toggle, Teleport Test, Leave Cell / Return, and
Disconnect After Delay
- Gameplay synchronization - Gameplay synchronization
- Authentication - Authentication
- Public matchmaking - Public matchmaking
+65 -10
View File
@@ -46,6 +46,7 @@ class DevServerWindow(QMainWindow):
host=self.server.host, host=self.server.host,
port=self.server.port, port=self.server.port,
log_callback=self._handle_fake_client_log, log_callback=self._handle_fake_client_log,
client_snapshot_provider=lambda: self.server.get_clients(),
) )
self.stat_labels: dict[str, QLabel] = {} self.stat_labels: dict[str, QLabel] = {}
@@ -173,6 +174,19 @@ class DevServerWindow(QMainWindow):
button_layout.addStretch() button_layout.addStretch()
layout.addLayout(button_layout) 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)
action_layout.addStretch()
layout.addLayout(action_layout)
self.fake_clients_table = QTableWidget() self.fake_clients_table = QTableWidget()
self.fake_clients_table.setColumnCount(6) self.fake_clients_table.setColumnCount(6)
self.fake_clients_table.setHorizontalHeaderLabels( self.fake_clients_table.setHorizontalHeaderLabels(
@@ -185,8 +199,8 @@ class DevServerWindow(QMainWindow):
layout.addWidget(self.fake_clients_table) layout.addWidget(self.fake_clients_table)
# TODO: Add future fake-client scripts: Walk Circle, Walk Square, # TODO: Add future fake-client scripts: Walk Circle, Walk Square,
# Jump Loop, Sneak Toggle, Teleport Test, Leave Cell / Return, and # Follow Player, Jump Loop, Sneak Toggle, Teleport Test,
# Disconnect After Delay. # Leave Cell / Return, and Disconnect After Delay.
return tab return tab
@@ -220,22 +234,63 @@ class DevServerWindow(QMainWindow):
self._refresh_fake_clients_table() self._refresh_fake_clients_table()
def _remove_selected_fake_client(self) -> None: def _remove_selected_fake_client(self) -> None:
selected_items = self.fake_clients_table.selectedItems() client_key = self._get_selected_fake_client_key()
if not selected_items: if client_key is None:
self._append_log("No fake client selected.") self._append_log("No fake client selected.")
return 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 _get_selected_fake_client_key_for_control(self) -> int | None:
client_key = self._get_selected_fake_client_key()
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() row = selected_items[0].row()
name_item = self.fake_clients_table.item(row, 0) name_item = self.fake_clients_table.item(row, 0)
if name_item is None: if name_item is None:
return return None
client_key = name_item.data(Qt.ItemDataRole.UserRole) client_key = name_item.data(Qt.ItemDataRole.UserRole)
if client_key is None: try:
return return int(client_key)
except (TypeError, ValueError):
self.fake_player_manager.remove_client(int(client_key)) return None
self._refresh_fake_clients_table()
def _handle_server_log(self, message: str) -> None: def _handle_server_log(self, message: str) -> None:
self.log_bridge.message_received.emit(message) self.log_bridge.message_received.emit(message)
+257 -14
View File
@@ -1,6 +1,7 @@
from __future__ import annotations from __future__ import annotations
import json import json
import math
import socket import socket
import threading import threading
import time import time
@@ -11,12 +12,18 @@ from typing import Any
HOST = "127.0.0.1" HOST = "127.0.0.1"
PORT = 7777 PORT = 7777
SCRIPT_IDLE = "Idle" SCRIPT_IDLE = "Idle"
SCRIPT_WALK_TO_PLAYER = "Walk To Player"
STATUS_CONNECTING = "Connecting" STATUS_CONNECTING = "Connecting"
STATUS_CONNECTED = "Connected" STATUS_CONNECTED = "Connected"
STATUS_DISCONNECTED = "Disconnected" STATUS_DISCONNECTED = "Disconnected"
STATUS_ERROR = "Error" STATUS_ERROR = "Error"
IDLE_SEND_INTERVAL_SECONDS = 1.0 IDLE_SEND_INTERVAL_SECONDS = 1.0
SOCKET_TIMEOUT_SECONDS = 0.2 WALK_TO_PLAYER_SEND_RATE_HZ = 10.0
WALK_TO_PLAYER_SPEED = 180.0
WALK_TO_PLAYER_STOP_DISTANCE = 100.0
WALK_TO_PLAYER_OFFSET_X = 150.0
NO_TARGET_LOG_INTERVAL_SECONDS = 5.0
SOCKET_TIMEOUT_SECONDS = 0.05
STOP_JOIN_TIMEOUT_SECONDS = 2.0 STOP_JOIN_TIMEOUT_SECONDS = 2.0
IDLE_CELL_ID = "0B000F99" IDLE_CELL_ID = "0B000F99"
@@ -26,6 +33,9 @@ IDLE_Y = -1344.0
IDLE_Z = 0.0 IDLE_Z = 0.0
IDLE_ANGLE_Z = 3.93 IDLE_ANGLE_Z = 3.93
ClientSnapshotProvider = Callable[[], list[dict[str, Any]]]
WalkTargetProvider = Callable[[int], dict[str, Any] | None]
class FakePlayerClient: class FakePlayerClient:
def __init__( def __init__(
@@ -36,6 +46,7 @@ class FakePlayerClient:
host: str = HOST, host: str = HOST,
port: int = PORT, port: int = PORT,
log_callback: Callable[[str], None] | None = None, log_callback: Callable[[str], None] | None = None,
walk_target_provider: WalkTargetProvider | None = None,
) -> None: ) -> None:
self.client_key = client_key self.client_key = client_key
self.name = name self.name = name
@@ -43,16 +54,23 @@ class FakePlayerClient:
self.port = port self.port = port
self.script = SCRIPT_IDLE self.script = SCRIPT_IDLE
self.cell_id = IDLE_CELL_ID self.cell_id = IDLE_CELL_ID
self.worldspace_id = IDLE_WORLDSPACE_ID
self.position = (IDLE_X, IDLE_Y, IDLE_Z) self.position = (IDLE_X, IDLE_Y, IDLE_Z)
self.angle_z = IDLE_ANGLE_Z
self._log_callback = log_callback self._log_callback = log_callback
self._walk_target_provider = walk_target_provider
self._lock = threading.RLock() self._lock = threading.RLock()
self._stop_event = threading.Event() self._stop_event = threading.Event()
self._script_changed_event = threading.Event()
self._thread: threading.Thread | None = None self._thread: threading.Thread | None = None
self._socket: socket.socket | None = None self._socket: socket.socket | None = None
self._player_id: int | None = None self._player_id: int | None = None
self._status = STATUS_DISCONNECTED self._status = STATUS_DISCONNECTED
self._last_error: str | None = None self._last_error: str | None = None
self._last_no_target_log_at = 0.0
self._last_target_player_id: int | None = None
self._reached_target_player_id: int | None = None
def start(self) -> None: def start(self) -> None:
with self._lock: with self._lock:
@@ -77,6 +95,12 @@ class FakePlayerClient:
if self._status != STATUS_ERROR: if self._status != STATUS_ERROR:
self._status = STATUS_DISCONNECTED self._status = STATUS_DISCONNECTED
def set_idle(self) -> None:
self._set_script(SCRIPT_IDLE)
def set_walk_to_player(self) -> None:
self._set_script(SCRIPT_WALK_TO_PLAYER)
def get_snapshot(self) -> dict[str, Any]: def get_snapshot(self) -> dict[str, Any]:
with self._lock: with self._lock:
return { return {
@@ -90,6 +114,16 @@ class FakePlayerClient:
"lastError": self._last_error, "lastError": self._last_error,
} }
def _set_script(self, script: str) -> None:
with self._lock:
self.script = script
self._last_no_target_log_at = 0.0
self._last_target_player_id = None
self._reached_target_player_id = None
self._script_changed_event.set()
self._log(f"{self.name} script set to {script}.")
def _run(self) -> None: def _run(self) -> None:
self._log(f"{self.name} connecting...") self._log(f"{self.name} connecting...")
try: try:
@@ -113,7 +147,9 @@ class FakePlayerClient:
def _receive_and_send_loop(self, connection: socket.socket) -> None: def _receive_and_send_loop(self, connection: socket.socket) -> None:
buffer = "" buffer = ""
next_transform_time = time.time() now = time.monotonic()
next_transform_time = now
last_transform_time = now
while not self._stop_event.is_set(): while not self._stop_event.is_set():
chunk: bytes | None chunk: bytes | None
@@ -138,10 +174,17 @@ class FakePlayerClient:
else: else:
break break
now = time.time() now = time.monotonic()
if self._script_changed_event.is_set():
self._script_changed_event.clear()
next_transform_time = now
last_transform_time = now
if self._is_connected() and now >= next_transform_time: if self._is_connected() and now >= next_transform_time:
self._send_idle_transform(connection) delta_seconds = max(0.0, now - last_transform_time)
next_transform_time = now + IDLE_SEND_INTERVAL_SECONDS interval_seconds = self._send_script_transform(connection, delta_seconds)
last_transform_time = now
next_transform_time = now + interval_seconds
def _handle_line(self, line: str) -> None: def _handle_line(self, line: str) -> None:
if not line: if not line:
@@ -173,27 +216,121 @@ class FakePlayerClient:
self._log(f"{self.name} assigned playerId {player_id}.") self._log(f"{self.name} assigned playerId {player_id}.")
def _send_idle_transform(self, connection: socket.socket) -> None: def _send_script_transform(self, connection: socket.socket, delta_seconds: float) -> float:
script = self._get_script()
if script == SCRIPT_WALK_TO_PLAYER:
return self._send_walk_to_player_transform(connection, delta_seconds)
self._send_transform(connection, is_moving=False, movement_speed=0.0)
return IDLE_SEND_INTERVAL_SECONDS
def _send_walk_to_player_transform(self, connection: socket.socket, delta_seconds: float) -> float:
target = self._get_walk_to_player_target()
if target is None:
self._log_no_target_if_needed()
self._send_transform(connection, is_moving=False, movement_speed=0.0)
return IDLE_SEND_INTERVAL_SECONDS
target_player_id = target["playerId"]
target_position = (target["x"], target["y"], target["z"])
walk_interval = 1.0 / WALK_TO_PLAYER_SEND_RATE_HZ
step_delta = max(delta_seconds, walk_interval)
with self._lock:
current_x, current_y, current_z = self.position
delta_x = target_position[0] - current_x
delta_y = target_position[1] - current_y
delta_z = target_position[2] - current_z
distance = math.sqrt((delta_x * delta_x) + (delta_y * delta_y) + (delta_z * delta_z))
if self._last_target_player_id != target_player_id:
self._last_target_player_id = target_player_id
self._reached_target_player_id = None
should_log_walking = True
else:
should_log_walking = False
self.cell_id = target["cellId"]
self.worldspace_id = target["worldspaceId"]
if distance <= WALK_TO_PLAYER_STOP_DISTANCE:
is_new_reached = self._reached_target_player_id != target_player_id
self._reached_target_player_id = target_player_id
is_moving = False
movement_speed = 0.0
else:
if self._reached_target_player_id == target_player_id:
should_log_walking = True
self._reached_target_player_id = None
max_step = WALK_TO_PLAYER_SPEED * step_delta
step = min(max_step, distance)
scale = step / distance
new_x = current_x + (delta_x * scale)
new_y = current_y + (delta_y * scale)
new_z = current_z + (delta_z * scale)
self.position = (new_x, new_y, new_z)
self.angle_z = math.atan2(delta_x, delta_y)
is_new_reached = False
is_moving = True
movement_speed = WALK_TO_PLAYER_SPEED
if should_log_walking:
self._log(f"{self.name} walking to player {target_player_id}.")
if is_new_reached:
self._log(f"{self.name} reached player {target_player_id}.")
self._send_transform(connection, is_moving=is_moving, movement_speed=movement_speed)
return walk_interval if is_moving else IDLE_SEND_INTERVAL_SECONDS
def _send_transform(self, connection: socket.socket, *, is_moving: bool, movement_speed: float) -> None:
with self._lock:
x, y, z = self.position
angle_z = self.angle_z
cell_id = self.cell_id
worldspace_id = self.worldspace_id
packet = { packet = {
"type": "transform", "type": "transform",
"x": IDLE_X, "x": x,
"y": IDLE_Y, "y": y,
"z": IDLE_Z, "z": z,
"angleZ": IDLE_ANGLE_Z, "angleZ": angle_z,
"movementType": "normal", "movementType": "normal",
"cellId": IDLE_CELL_ID, "cellId": cell_id,
"worldspaceId": IDLE_WORLDSPACE_ID, "worldspaceId": worldspace_id,
"clientTime": time.time(), "clientTime": time.time(),
"isMoving": False, "isMoving": is_moving,
"isSprinting": False, "isSprinting": False,
"isSneaking": False, "isSneaking": False,
"isJumping": False, "isJumping": False,
"weaponDrawn": False, "weaponDrawn": False,
"movementSpeed": 0.0, "movementSpeed": movement_speed,
} }
encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n" encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n"
connection.sendall(encoded) connection.sendall(encoded)
def _get_script(self) -> str:
with self._lock:
return self.script
def _get_walk_to_player_target(self) -> dict[str, Any] | None:
if self._walk_target_provider is None:
return None
return self._walk_target_provider(self.client_key)
def _log_no_target_if_needed(self) -> None:
now = time.monotonic()
with self._lock:
if now - self._last_no_target_log_at < NO_TARGET_LOG_INTERVAL_SECONDS:
return
self._last_no_target_log_at = now
self._last_target_player_id = None
self._reached_target_player_id = None
self._log("No real player target available for Walk To Player.")
def _is_connected(self) -> bool: def _is_connected(self) -> bool:
with self._lock: with self._lock:
return self._status == STATUS_CONNECTED return self._status == STATUS_CONNECTED
@@ -235,10 +372,12 @@ class FakePlayerManager:
host: str = HOST, host: str = HOST,
port: int = PORT, port: int = PORT,
log_callback: Callable[[str], None] | None = None, log_callback: Callable[[str], None] | None = None,
client_snapshot_provider: ClientSnapshotProvider | None = None,
) -> None: ) -> None:
self.host = host self.host = host
self.port = port self.port = port
self._log_callback = log_callback self._log_callback = log_callback
self._client_snapshot_provider = client_snapshot_provider
self._lock = threading.RLock() self._lock = threading.RLock()
self._next_client_key = 1 self._next_client_key = 1
self._clients: dict[int, FakePlayerClient] = {} self._clients: dict[int, FakePlayerClient] = {}
@@ -253,6 +392,7 @@ class FakePlayerManager:
host=self.host, host=self.host,
port=self.port, port=self.port,
log_callback=self._log, log_callback=self._log,
walk_target_provider=self.get_walk_to_player_target,
) )
self._clients[client_key] = client self._clients[client_key] = client
@@ -266,6 +406,22 @@ class FakePlayerManager:
if client is not None: if client is not None:
client.stop() client.stop()
def set_client_idle(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.set_idle()
return True
def set_client_walk_to_player(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.set_walk_to_player()
return True
def stop_all(self) -> None: def stop_all(self) -> None:
with self._lock: with self._lock:
clients = list(self._clients.values()) clients = list(self._clients.values())
@@ -280,6 +436,93 @@ class FakePlayerManager:
return [client.get_snapshot() for client in clients] return [client.get_snapshot() for client in clients]
def get_walk_to_player_target(self, client_key: int) -> dict[str, Any] | None:
if self._client_snapshot_provider is None:
return None
try:
server_clients = self._client_snapshot_provider()
except Exception as error:
self._log(f"Could not read server client snapshots: {error}")
return None
fake_player_ids = self._get_fake_player_ids()
target_candidates: list[dict[str, Any]] = []
for client in server_clients:
target = self._parse_walk_target(client)
if target is None or target["playerId"] in fake_player_ids:
continue
target_candidates.append(target)
if not target_candidates:
return None
target_candidates.sort(key=lambda target: target["playerId"])
return target_candidates[0]
def _get_client(self, client_key: int) -> FakePlayerClient | None:
with self._lock:
return self._clients.get(client_key)
def _get_fake_player_ids(self) -> set[int]:
with self._lock:
clients = list(self._clients.values())
fake_player_ids: set[int] = set()
for client in clients:
snapshot = client.get_snapshot()
player_id = self._get_optional_int(snapshot.get("playerId"))
if player_id is not None:
fake_player_ids.add(player_id)
return fake_player_ids
@classmethod
def _parse_walk_target(cls, client: dict[str, Any]) -> dict[str, Any] | None:
player_id = cls._get_optional_int(client.get("playerId"))
last_transform = client.get("lastTransform")
if player_id is None or not isinstance(last_transform, dict):
return None
x = cls._get_finite_float(last_transform.get("x"))
y = cls._get_finite_float(last_transform.get("y"))
z = cls._get_finite_float(last_transform.get("z"))
cell_id = last_transform.get("cellId")
if x is None or y is None or z is None or not cell_id:
return None
worldspace_id = last_transform.get("worldspaceId", "")
if worldspace_id is None:
worldspace_id = ""
return {
"playerId": player_id,
"x": x + WALK_TO_PLAYER_OFFSET_X,
"y": y,
"z": z,
"cellId": str(cell_id),
"worldspaceId": str(worldspace_id),
}
@staticmethod
def _get_optional_int(value: Any) -> int | None:
try:
return int(value)
except (TypeError, ValueError):
return None
@staticmethod
def _get_finite_float(value: Any) -> float | None:
try:
parsed_value = float(value)
except (TypeError, ValueError):
return None
if not math.isfinite(parsed_value):
return None
return parsed_value
def _log(self, message: str) -> None: def _log(self, message: str) -> None:
if self._log_callback is None: if self._log_callback is None:
return return