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 Fallout 4 Together 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 Fallout 4 Together server at {HOST}:{PORT}. Is server.py running?") except KeyboardInterrupt: log("\nFake client stopped.") except OSError as error: log(f"Disconnected from server: {error}")