Commonwealth-Online/Commonwealth-Online-Public
2
0
Fork 0
Code Pull Requests Activity
Files
ad7f2878992921494dc7b2f56c4a5c72522e6907
Commonwealth-Online-Public/server/fake_player.py
T
andrew 5ba440ab5d Add crouch state and fix jump sync 
Add isCrouching to network, state and animation systems; enable proxy jump syncing with a jump-hold debounce and update animation descriptor and server tooling.

Key changes:
- Networking: F4TNetworking.h/cpp: added a_isCrouching, include "isCrouching" in transform packets and parsing, and logging.
- Remote state: F4TRemotePlayerState.h: added isCrouching field.
- Proxy animation: F4TProxyAnimationSync.*: added crouch fields, enabled jump sync, implemented ApplyJumpHoldState and jump hold timing, wrote crouch/jump bools into descriptors when available, updated event firing and logs.
- Animation descriptor: F4AnimationDescriptor.cpp: added isJumping and isCrouching to bool variable list.
- Local detection: main.cpp: added isCrouching in movement state, stubbed GetCrouchState(), adjusted jump hold timing and included crouch in outgoing transform calls.
- Server/dev tooling: server/dev_server_app.py, fake_client.py, fake_player.py, README.md: added crouch reporting, toggle UI/control and fake client support for crouch, and updated docs/logging.
- Docs: docs/dev-log.md updated with notes on jump/crouch fixes and testing.

Rationale: ensure remote players can report crouch state and make jump animations transition reliably by debouncing jump state on proxies and writing available graph bools; also provide dev-server controls and fake-client support for testing.
2026-06-05 14:22:17 +12:00

858 lines
28 KiB
Python
Raw Blame History

from __future__ import annotations
import json
import math
import socket
import threading
import time
from collections.abc import Callable
from typing import Any
HOST = "127.0.0.1"
PORT = 7777
SCRIPT_IDLE = "Idle"
SCRIPT_WALK_TO_PLAYER = "Walk To Player"
SCRIPT_WALK_CIRCLE = "Walk Circle"
SCRIPT_LEFT_CELL = "Left Cell"
STATUS_CONNECTING = "Connecting"
STATUS_CONNECTED = "Connected"
STATUS_DISCONNECTED = "Disconnected"
STATUS_ERROR = "Error"
IDLE_SEND_INTERVAL_SECONDS = 1.0
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
WALK_CIRCLE_RADIUS = 300.0
WALK_CIRCLE_SPEED = 180.0
WALK_CIRCLE_SEND_RATE_HZ = 10.0
JUMP_DURATION_SECONDS = 0.8
JUMP_HEIGHT = 120.0
JUMP_SEND_RATE_HZ = 10.0
NO_TARGET_LOG_INTERVAL_SECONDS = 5.0
SOCKET_TIMEOUT_SECONDS = 0.05
STOP_JOIN_TIMEOUT_SECONDS = 2.0
IDLE_CELL_ID = "0B000F99"
IDLE_WORLDSPACE_ID = ""
LEFT_CELL_ID = "DEADBEEF"
IDLE_X = -1792.0
IDLE_Y = -1344.0
IDLE_Z = 0.0
IDLE_ANGLE_Z = 3.93
TELEPORT_TEST_OFFSET_X = 900.0
TELEPORT_TEST_OFFSET_Y = 450.0
ClientSnapshotProvider = Callable[[], list[dict[str, Any]]]
WalkTargetProvider = Callable[[int], dict[str, Any] | None]
def map_movement_speed_to_graph_speed(movement_speed: float, *, is_sprinting: bool = False) -> float:
"""Match plugin MapMovementSpeedToGraphSpeed.
FO4 graph "Speed" shares the same world-units/sec scale as movementSpeed, so the
mapping is ~1:1 (clamped). Confirmed from local-player graph debug: run movementSpeed
~306 lines up with graph Speed ~296-373.
"""
if movement_speed < 1.0:
return 0.0
graph_speed = movement_speed
if is_sprinting:
graph_speed = max(graph_speed, 350.0)
return min(graph_speed, 500.0)
class FakePlayerClient:
def __init__(
self,
client_key: int,
name: str,
*,
host: str = HOST,
port: int = PORT,
log_callback: Callable[[str], None] | None = None,
walk_target_provider: WalkTargetProvider | None = None,
) -> None:
self.client_key = client_key
self.name = name
self.host = host
self.port = port
self.script = SCRIPT_IDLE
self.cell_id = IDLE_CELL_ID
self.worldspace_id = IDLE_WORLDSPACE_ID
self.position = (IDLE_X, IDLE_Y, IDLE_Z)
self.angle_z = IDLE_ANGLE_Z
self.is_sneaking = False
self.is_jumping = False
self.is_crouching = False
self._log_callback = log_callback
self._walk_target_provider = walk_target_provider
self._lock = threading.RLock()
self._send_lock = threading.Lock()
self._stop_event = threading.Event()
self._script_changed_event = threading.Event()
self._thread: threading.Thread | None = None
self._socket: socket.socket | None = None
self._player_id: int | None = None
self._status = STATUS_DISCONNECTED
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
self._walk_circle_center = self.position
self._walk_circle_phase = 0.0
self._one_shot_active = False
self._jump_thread: threading.Thread | None = None
self._teleport_count = 0
def start(self) -> None:
with self._lock:
if self._thread is not None and self._thread.is_alive():
return
self._stop_event.clear()
self._status = STATUS_CONNECTING
self._last_error = None
self._thread = threading.Thread(target=self._run, name=self.name, daemon=True)
self._thread.start()
def stop(self) -> None:
self._stop_event.set()
self._close_socket()
thread = self._thread
if thread is not None and thread is not threading.current_thread():
thread.join(timeout=STOP_JOIN_TIMEOUT_SECONDS)
jump_thread = self._jump_thread
if jump_thread is not None and jump_thread is not threading.current_thread():
jump_thread.join(timeout=STOP_JOIN_TIMEOUT_SECONDS)
with self._lock:
if self._status != STATUS_ERROR:
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 set_walk_circle(self) -> None:
with self._lock:
self.cell_id = IDLE_CELL_ID
self.worldspace_id = IDLE_WORLDSPACE_ID
self._walk_circle_center = self.position
self._walk_circle_phase = 0.0
self._set_script(SCRIPT_WALK_CIRCLE)
def trigger_jump_once(self) -> None:
with self._lock:
if self._one_shot_active:
return
self._one_shot_active = True
previous_script = self.script
thread = threading.Thread(
target=self._run_jump_once,
args=(previous_script,),
name=f"{self.name} Jump Once",
daemon=True,
)
self._jump_thread = thread
self._log(f"{self.name} triggered Jump Once.")
thread.start()
def toggle_sneak(self) -> bool:
with self._lock:
self.is_sneaking = not self.is_sneaking
is_sneaking = self.is_sneaking
self._script_changed_event.set()
self._log(f"{self.name} toggled sneak: {str(is_sneaking).lower()}.")
return is_sneaking
def toggle_crouch(self) -> bool:
with self._lock:
self.is_crouching = not self.is_crouching
is_crouching = self.is_crouching
self._script_changed_event.set()
self._log(f"{self.name} toggled crouch: {str(is_crouching).lower()}.")
return is_crouching
def trigger_leave_cell(self) -> None:
with self._lock:
self.script = SCRIPT_LEFT_CELL
self.cell_id = LEFT_CELL_ID
self.worldspace_id = IDLE_WORLDSPACE_ID
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._send_immediate_transform(
movement_type="cell_change",
is_moving=False,
movement_speed=0.0,
)
self._log(f"{self.name} left test cell.")
def trigger_return_to_cell(self) -> None:
with self._lock:
self.script = SCRIPT_IDLE
self.cell_id = IDLE_CELL_ID
self.worldspace_id = IDLE_WORLDSPACE_ID
self.position = (IDLE_X, IDLE_Y, IDLE_Z)
self.angle_z = IDLE_ANGLE_Z
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._send_immediate_transform(
movement_type="cell_change",
is_moving=False,
movement_speed=0.0,
)
self._log(f"{self.name} returned to test cell.")
def trigger_teleport_test(self) -> None:
with self._lock:
self.script = SCRIPT_IDLE
self.cell_id = IDLE_CELL_ID
self.worldspace_id = IDLE_WORLDSPACE_ID
current_x, current_y, current_z = self.position
direction = -1.0 if self._teleport_count % 2 else 1.0
self._teleport_count += 1
self.position = (
current_x + (TELEPORT_TEST_OFFSET_X * direction),
current_y + (TELEPORT_TEST_OFFSET_Y * direction),
current_z,
)
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._send_immediate_transform(
movement_type="teleport",
is_moving=False,
movement_speed=0.0,
)
self._log(f"{self.name} triggered Teleport Test.")
def get_snapshot(self) -> dict[str, Any]:
with self._lock:
return {
"clientKey": self.client_key,
"name": self.name,
"playerId": self._player_id,
"status": self._status,
"script": self.script,
"cellId": self.cell_id,
"position": self.position,
"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:
self._log(f"{self.name} connecting...")
try:
connection = socket.create_connection((self.host, self.port), timeout=5.0)
connection.settimeout(SOCKET_TIMEOUT_SECONDS)
with self._lock:
self._socket = connection
self._receive_and_send_loop(connection)
except (ConnectionRefusedError, TimeoutError, OSError) as error:
if not self._stop_event.is_set():
self._set_error(str(error))
self._log(f"{self.name} error: {error}")
finally:
self._close_socket()
with self._lock:
if self._status != STATUS_ERROR:
self._status = STATUS_DISCONNECTED
self._log(f"{self.name} disconnected.")
def _receive_and_send_loop(self, connection: socket.socket) -> None:
buffer = ""
now = time.monotonic()
next_transform_time = now
last_transform_time = now
while not self._stop_event.is_set():
chunk: bytes | None
try:
chunk = connection.recv(4096)
except socket.timeout:
chunk = None
except ConnectionResetError:
break
except OSError:
if not self._stop_event.is_set():
raise
break
if chunk is None:
pass
elif chunk:
buffer += chunk.decode("utf-8", errors="replace")
while "\n" in buffer:
line, buffer = buffer.split("\n", 1)
self._handle_line(line.strip())
else:
break
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:
delta_seconds = max(0.0, now - last_transform_time)
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:
if not line:
return
try:
packet = json.loads(line)
except json.JSONDecodeError as error:
self._log(f"{self.name} received invalid JSON: {error}: {line}")
return
if not isinstance(packet, dict):
self._log(f"{self.name} received non-object packet: {packet}")
return
if packet.get("type") == "welcome":
self._handle_welcome(packet)
def _handle_welcome(self, packet: dict[str, Any]) -> None:
try:
player_id = int(packet["playerId"])
except (KeyError, TypeError, ValueError):
self._log(f"{self.name} received malformed welcome packet: {packet}")
return
with self._lock:
self._player_id = player_id
self._status = STATUS_CONNECTED
self._log(f"{self.name} assigned playerId {player_id}.")
def _send_script_transform(self, connection: socket.socket, delta_seconds: float) -> float:
script = self._get_script()
if self._is_one_shot_active():
return SOCKET_TIMEOUT_SECONDS
if script == SCRIPT_WALK_TO_PLAYER:
return self._send_walk_to_player_transform(connection, delta_seconds)
if script == SCRIPT_WALK_CIRCLE:
return self._send_walk_circle_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_walk_circle_transform(self, connection: socket.socket, delta_seconds: float) -> float:
walk_interval = 1.0 / WALK_CIRCLE_SEND_RATE_HZ
step_delta = max(delta_seconds, walk_interval)
angular_speed = WALK_CIRCLE_SPEED / WALK_CIRCLE_RADIUS
with self._lock:
self.cell_id = IDLE_CELL_ID
self.worldspace_id = IDLE_WORLDSPACE_ID
center_x, center_y, center_z = self._walk_circle_center
self._walk_circle_phase += angular_speed * step_delta
phase = self._walk_circle_phase
new_x = center_x + (WALK_CIRCLE_RADIUS * math.cos(phase))
new_y = center_y + (WALK_CIRCLE_RADIUS * math.sin(phase))
new_z = center_z
tangent_x = -math.sin(phase)
tangent_y = math.cos(phase)
self.position = (new_x, new_y, new_z)
self.angle_z = math.atan2(tangent_x, tangent_y)
self._send_transform(connection, is_moving=True, movement_speed=WALK_CIRCLE_SPEED)
return walk_interval
def _run_jump_once(self, previous_script: str) -> None:
interval_seconds = 1.0 / JUMP_SEND_RATE_HZ
steps = max(2, int(JUMP_DURATION_SECONDS * JUMP_SEND_RATE_HZ))
with self._lock:
ground_x, ground_y, ground_z = self.position
try:
for step_index in range(steps + 1):
if self._stop_event.is_set():
return
progress = step_index / steps
height = JUMP_HEIGHT * (1.0 - abs((2.0 * progress) - 1.0))
with self._lock:
self.position = (ground_x, ground_y, ground_z + height)
self.is_jumping = True
self._send_immediate_transform(
movement_type="normal",
is_moving=False,
movement_speed=0.0,
is_jumping=True,
)
time.sleep(interval_seconds)
finally:
with self._lock:
self.position = (ground_x, ground_y, ground_z)
self.is_jumping = False
self._one_shot_active = False
if self.script == previous_script and previous_script not in (
SCRIPT_WALK_TO_PLAYER,
SCRIPT_WALK_CIRCLE,
SCRIPT_LEFT_CELL,
):
self.script = SCRIPT_IDLE
self._script_changed_event.set()
self._send_immediate_transform(
movement_type="normal",
is_moving=False,
movement_speed=0.0,
is_jumping=False,
)
def _send_immediate_transform(
self,
*,
movement_type: str,
is_moving: bool,
movement_speed: float,
is_jumping: bool | None = None,
is_crouching: bool | None = None,
) -> None:
with self._lock:
connection = self._socket
is_connected = self._status == STATUS_CONNECTED
if connection is None or not is_connected:
return
try:
self._send_transform(
connection,
movement_type=movement_type,
is_moving=is_moving,
movement_speed=movement_speed,
is_jumping=is_jumping,
is_crouching=is_crouching,
)
except OSError as error:
if not self._stop_event.is_set():
self._set_error(str(error))
self._log(f"{self.name} error: {error}")
def _send_transform(
self,
connection: socket.socket,
*,
is_moving: bool,
movement_speed: float,
movement_type: str = "normal",
is_jumping: bool | None = None,
is_crouching: bool | None = None,
) -> None:
with self._lock:
x, y, z = self.position
angle_z = self.angle_z
cell_id = self.cell_id
worldspace_id = self.worldspace_id
is_sneaking = self.is_sneaking
jumping = self.is_jumping if is_jumping is None else is_jumping
crouching = self.is_crouching if is_crouching is None else is_crouching
packet = {
"type": "transform",
"x": x,
"y": y,
"z": z,
"angleZ": angle_z,
"movementType": movement_type,
"cellId": cell_id,
"worldspaceId": worldspace_id,
"clientTime": time.time(),
"isMoving": is_moving,
"isSprinting": False,
"isSneaking": is_sneaking,
"isJumping": jumping,
"isCrouching": crouching,
"weaponDrawn": False,
"movementSpeed": movement_speed,
}
if is_moving and movement_speed >= 1.0:
packet["animationGraphSpeed"] = round(
map_movement_speed_to_graph_speed(movement_speed),
1,
)
encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n"
with self._send_lock:
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:
with self._lock:
return self._status == STATUS_CONNECTED
def _is_one_shot_active(self) -> bool:
with self._lock:
return self._one_shot_active
def _set_error(self, message: str) -> None:
with self._lock:
self._status = STATUS_ERROR
self._last_error = message
def _close_socket(self) -> None:
with self._lock:
connection = self._socket
self._socket = None
if connection is None:
return
try:
connection.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
connection.close()
except OSError:
pass
def _log(self, message: str) -> None:
if self._log_callback is None:
return
self._log_callback(message)
class FakePlayerManager:
def __init__(
self,
*,
host: str = HOST,
port: int = PORT,
log_callback: Callable[[str], None] | None = None,
client_snapshot_provider: ClientSnapshotProvider | None = None,
) -> None:
self.host = host
self.port = port
self._log_callback = log_callback
self._client_snapshot_provider = client_snapshot_provider
self._lock = threading.RLock()
self._next_client_key = 1
self._clients: dict[int, FakePlayerClient] = {}
def add_client(self) -> FakePlayerClient:
with self._lock:
client_key = self._next_client_key
self._next_client_key += 1
client = FakePlayerClient(
client_key,
f"Fake Client {client_key}",
host=self.host,
port=self.port,
log_callback=self._log,
walk_target_provider=self.get_walk_to_player_target,
)
self._clients[client_key] = client
client.start()
return client
def remove_client(self, client_key: int) -> None:
with self._lock:
client = self._clients.pop(client_key, None)
if client is not None:
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 set_client_walk_circle(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.set_walk_circle()
return True
def trigger_client_jump_once(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.trigger_jump_once()
return True
def toggle_client_sneak(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.toggle_sneak()
return True
def toggle_client_crouch(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.toggle_crouch()
return True
def trigger_client_leave_cell(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.trigger_leave_cell()
return True
def trigger_client_return_to_cell(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.trigger_return_to_cell()
return True
def trigger_client_teleport_test(self, client_key: int) -> bool:
client = self._get_client(client_key)
if client is None:
return False
client.trigger_teleport_test()
return True
def stop_all(self) -> None:
with self._lock:
clients = list(self._clients.values())
self._clients.clear()
for client in clients:
client.stop()
def get_snapshots(self) -> list[dict[str, Any]]:
with self._lock:
clients = list(self._clients.values())
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:
if self._log_callback is None:
return
self._log_callback(f"[FakeClient] {message}")
View Git Blame Copy Permalink