Commonwealth-Online-Public/server/fake_client.py

from __future__ import annotations

import json
import math
import socket
import time
from typing import Any


HOST = "127.0.0.1"
PORT = 7777

remote_players: dict[int, dict[str, Any]] = {}


def log(message: str) -> None:
    print(message, flush=True)


def warn(message: str) -> None:
    log(f"Warning: {message}")


def get_optional_uint32(packet: dict[str, Any], field_name: str, default: int = 0) -> int:
    """Get optional unsigned 32-bit integer field."""
    value = packet.get(field_name, default)
    try:
        parsed_value = int(value)
        if 0 <= parsed_value <= 0xFFFFFFFF:
            return parsed_value
    except (TypeError, ValueError):
        pass
    return default


def get_optional_bool(packet: dict[str, Any], field_name: str, default: bool = False) -> bool:
    value = packet.get(field_name, default)
    if isinstance(value, bool):
        return value
    return default


def get_optional_float(packet: dict[str, Any], field_name: str, default: float = 0.0) -> float:
    value = packet.get(field_name, default)
    try:
        parsed_value = float(value)
    except (TypeError, ValueError):
        return default

    if not math.isfinite(parsed_value) or parsed_value < 0.0:
        return default
    return parsed_value


def update_remote_player_state(packet: dict[str, Any]) -> None:
    required_fields = (
        "type",
        "playerId",
        "x",
        "y",
        "z",
        "angleZ",
        "cellId",
    )
    missing_fields = [field_name for field_name in required_fields if field_name not in packet]
    if missing_fields:
        warn(f"Ignoring transform packet missing required field {missing_fields[0]}: {packet}")
        return

    try:
        player_id = int(packet["playerId"])
        x = float(packet["x"])
        y = float(packet["y"])
        z = float(packet["z"])
        angle_z = float(packet["angleZ"])
    except (TypeError, ValueError) as error:
        warn(f"Ignoring transform packet with invalid numeric fields: {error}: {packet}")
        return

    remote_players[player_id] = {
        "playerId": player_id,
        "x": x,
        "y": y,
        "z": z,
        "angleZ": angle_z,
        "movementType": packet.get("movementType", "normal"),
        "cellId": packet["cellId"],
        "worldspaceId": packet.get("worldspaceId", ""),
        "clientTime": packet.get("clientTime", 0.0),
        "serverTime": packet.get("serverTime", 0.0),
        "isMoving": get_optional_bool(packet, "isMoving"),
        "isSprinting": get_optional_bool(packet, "isSprinting"),
        "isSneaking": get_optional_bool(packet, "isSneaking"),
        "isJumping": get_optional_bool(packet, "isJumping"),
        "isCrouching": get_optional_bool(packet, "isCrouching"),
        "weaponDrawn": get_optional_bool(packet, "weaponDrawn"),
        "movementSpeed": get_optional_float(packet, "movementSpeed"),
        "actorStateFlags1": get_optional_uint32(packet, "actorStateFlags1"),
        "actorStateFlags2": get_optional_uint32(packet, "actorStateFlags2"),
        "lastReceivedLocalTime": time.time(),
    }

    log(f"Updated remote player {player_id}")
    print_remote_player(remote_players[player_id])


def print_remote_player(player: dict[str, Any]) -> None:
    client_time = player["clientTime"] if player["clientTime"] is not None else "missing from packet"
    server_time = player["serverTime"] if player["serverTime"] is not None else "missing from packet"

    log(
        "\n".join(
            [
                f"Position: X={player['x']:.2f}, Y={player['y']:.2f}, Z={player['z']:.2f}",
                f"AngleZ: {player['angleZ']:.2f}",
                f"Movement Type: {player['movementType']}",
                (
                    "Movement State: "
                    f"moving={player['isMoving']}, sprinting={player['isSprinting']}, "
                    f"sneaking={player['isSneaking']}, jumping={player['isJumping']}, "
                    f"crouching={player['isCrouching']}, weaponDrawn={player['weaponDrawn']}, speed={player['movementSpeed']:.1f}"
                ),
                f"Actor State: flags1={player['actorStateFlags1']:08X}, flags2={player['actorStateFlags2']:08X}",
                f"Cell: {player['cellId']}",
                f"Worldspace: {player['worldspaceId']}",
                f"Client Time: {client_time}",
                f"Last Server Time: {server_time}",
                f"Last Received Local Time: {player['lastReceivedLocalTime']}",
            ]
        )
    )


def print_remote_players() -> None:
    if not remote_players:
        log("No remote players currently tracked.")
        return

    log("Known remote players:")
    for player_id in sorted(remote_players):
        player = remote_players[player_id]
        log(
            f"- Player {player_id}: "
            f"pos=({player['x']:.2f}, {player['y']:.2f}, {player['z']:.2f}), "
            f"angleZ={player['angleZ']:.2f}, movementType={player['movementType']}, "
            f"moving={player['isMoving']}, sprinting={player['isSprinting']}, "
            f"sneaking={player['isSneaking']}, jumping={player['isJumping']}, "
            f"crouching={player['isCrouching']}, weaponDrawn={player['weaponDrawn']}, speed={player['movementSpeed']:.1f}, "
            f"flags1={player['actorStateFlags1']:08X}, flags2={player['actorStateFlags2']:08X}, "
            f"cell={player['cellId']}, worldspace={player['worldspaceId']}, "
            f"serverTime={player['serverTime']}"
        )


def handle_disconnect_packet(packet: dict[str, Any]) -> None:
    player_id = packet.get("playerId")
    if player_id is None:
        warn(f"Ignoring disconnect packet without playerId: {packet}")
        return

    try:
        player_id = int(player_id)
    except (TypeError, ValueError):
        warn(f"Ignoring disconnect packet with invalid playerId: {packet}")
        return

    removed_player = remote_players.pop(player_id, None)
    if removed_player is None:
        log(f"Received disconnect for unknown remote player {player_id}.")
        return

    log(f"Removed remote player {player_id} after disconnect.")
    print_remote_players()


def handle_packet(packet: dict[str, Any]) -> None:
    packet_type = packet.get("type")
    if packet_type == "welcome":
        log(f"Welcome packet: {packet}")
    elif packet_type == "transform":
        log(f"Received transform: {packet}")
        update_remote_player_state(packet)
    elif packet_type == "disconnect":
        handle_disconnect_packet(packet)
    else:
        log(f"Received packet: {packet}")


def handle_line(line: str) -> None:
    if not line:
        return

    try:
        packet = json.loads(line)
    except json.JSONDecodeError as error:
        log(f"Invalid JSON from server: {error}: {line}")
        return

    if not isinstance(packet, dict):
        log(f"Invalid packet from server: expected JSON object: {packet}")
        return

    handle_packet(packet)


def main() -> None:
    # This fake client lets us verify server broadcast behavior before trying to
    # run and coordinate a second Fallout 4/F4SE instance.
    with socket.create_connection((HOST, PORT)) as connection:
        log(f"Connected to Commonwealth Online server at {HOST}:{PORT}")

        buffer = ""
        while True:
            chunk = connection.recv(4096)
            if not chunk:
                log("Server disconnected.")
                break

            buffer += chunk.decode("utf-8", errors="replace")
            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                handle_line(line.strip())


if __name__ == "__main__":
    try:
        main()
    except ConnectionRefusedError:
        log(f"Could not connect to Commonwealth Online server at {HOST}:{PORT}. Is server.py running?")
    except KeyboardInterrupt:
        log("\nFake client stopped.")
    except OSError as error:
        log(f"Disconnected from server: {error}")