Commonwealth-Online-Public/server/fake_player.py

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._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 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,
    ) -> 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,
            )
        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,
    ) -> 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

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