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}")