andrew
d328bfb647
Introduce complexionFormId (TXST) into RemoteAppearance and the appearance protocol so complexion textures are captured and serialized. Updates: header/state, Networking JSON parse/serialize, proxy actor comparisons, main capture logic, protocol docs, fake client/player samples, changelog and dev-log. Capture reads PlayerCharacter::complexion and serializes it, but applying to proxies is intentionally a no-op because TESNPC proxies lack a complexion field (documented in comments/dev-log). Files: plugin/include/F4TRemotePlayerState.h, plugin/src/*, protocol/packets.md, server/fake_client.py, server/fake_player.py, changelog.md, docs/dev-log.md.
461 lines
17 KiB
Python
461 lines
17 KiB
Python
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]] = {}
|
|
host_world_state: dict[str, Any] = {}
|
|
world_state_host_player_id: int | None = None
|
|
|
|
|
|
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 get_optional_game_time(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):
|
|
return default
|
|
return parsed_value
|
|
|
|
|
|
def get_optional_equipped_items(packet: dict[str, Any]) -> list[dict[str, str]]:
|
|
value = packet.get("equippedItems", [])
|
|
if not isinstance(value, list):
|
|
return []
|
|
|
|
equipped_items: list[dict[str, str]] = []
|
|
for item in value:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
|
|
slot = item.get("slot")
|
|
form_id = item.get("formId", "")
|
|
if isinstance(slot, str) and isinstance(form_id, str):
|
|
equipped_items.append({"slot": slot, "formId": form_id})
|
|
|
|
return equipped_items
|
|
|
|
|
|
def get_optional_appearance(packet: dict[str, Any]) -> dict[str, Any] | None:
|
|
value = packet.get("appearance")
|
|
if not isinstance(value, dict):
|
|
return None
|
|
|
|
appearance: dict[str, Any] = {
|
|
"version": get_optional_uint32(value, "version", 1),
|
|
"raceFormId": value.get("raceFormId", ""),
|
|
"height": get_optional_float(value, "height", 1.0),
|
|
"morphWeight": value.get("morphWeight", {}),
|
|
"bodyTintColor": value.get("bodyTintColor", {}),
|
|
"hairColorFormId": value.get("hairColorFormId", ""),
|
|
"facialHairColorFormId": value.get("facialHairColorFormId", ""),
|
|
"complexionFormId": value.get("complexionFormId", ""),
|
|
"headParts": [],
|
|
"morphs": [],
|
|
"morphRegions": [],
|
|
"facialBoneMorphs": [],
|
|
"tints": [],
|
|
}
|
|
|
|
head_parts = value.get("headParts", [])
|
|
if isinstance(head_parts, list):
|
|
appearance["headParts"] = [form_id for form_id in head_parts if isinstance(form_id, str)]
|
|
|
|
morphs = value.get("morphs", [])
|
|
if isinstance(morphs, list):
|
|
appearance["morphs"] = [
|
|
{"id": morph.get("id", ""), "value": get_optional_float(morph, "value")}
|
|
for morph in morphs
|
|
if isinstance(morph, dict) and isinstance(morph.get("id", ""), str)
|
|
]
|
|
|
|
# Added in appearance version 2; older senders omit these and the lists stay empty.
|
|
morph_regions = value.get("morphRegions", [])
|
|
if isinstance(morph_regions, list):
|
|
appearance["morphRegions"] = [
|
|
get_optional_float({"v": entry}, "v")
|
|
for entry in morph_regions
|
|
if isinstance(entry, (int, float))
|
|
]
|
|
|
|
facial_bone_morphs = value.get("facialBoneMorphs", [])
|
|
if isinstance(facial_bone_morphs, list):
|
|
appearance["facialBoneMorphs"] = [
|
|
{
|
|
"id": morph.get("id", ""),
|
|
"position": _read_float3(morph.get("position")),
|
|
"rotation": _read_float3(morph.get("rotation")),
|
|
"scale": _read_float3(morph.get("scale"), default=1.0),
|
|
}
|
|
for morph in facial_bone_morphs
|
|
if isinstance(morph, dict) and isinstance(morph.get("id", ""), str)
|
|
]
|
|
|
|
# Added in appearance version 3; palette-only color/swatch are preserved when present.
|
|
tints = value.get("tints", [])
|
|
if isinstance(tints, list):
|
|
parsed_tints: list[dict[str, Any]] = []
|
|
for tint in tints:
|
|
if not isinstance(tint, dict) or not isinstance(tint.get("id"), (int, float)):
|
|
continue
|
|
entry: dict[str, Any] = {
|
|
"id": int(tint.get("id", 0)),
|
|
"type": int(tint["type"]) if isinstance(tint.get("type"), (int, float)) else 0,
|
|
"value": int(tint["value"]) if isinstance(tint.get("value"), (int, float)) else 0,
|
|
}
|
|
color = tint.get("color")
|
|
if isinstance(color, str) and color:
|
|
entry["color"] = color
|
|
if isinstance(tint.get("swatch"), (int, float)):
|
|
entry["swatch"] = int(tint["swatch"])
|
|
parsed_tints.append(entry)
|
|
appearance["tints"] = parsed_tints
|
|
|
|
return appearance
|
|
|
|
|
|
def _read_float3(value: Any, default: float = 0.0) -> list[float]:
|
|
result = [default, default, default]
|
|
if isinstance(value, list):
|
|
for index in range(min(len(value), 3)):
|
|
element = value[index]
|
|
if isinstance(element, (int, float)):
|
|
result[index] = float(element)
|
|
return result
|
|
|
|
|
|
def update_host_world_state(packet: dict[str, Any]) -> None:
|
|
global host_world_state
|
|
|
|
host_world_state = {
|
|
"gameHour": get_optional_game_time(packet, "gameHour"),
|
|
"gameDaysPassed": get_optional_game_time(packet, "gameDaysPassed"),
|
|
"weatherFormId": packet.get("weatherFormId", ""),
|
|
"timeSync": packet.get("timeSync", False),
|
|
"clientTime": packet.get("clientTime", 0.0),
|
|
"serverTime": packet.get("serverTime", 0.0),
|
|
"lastReceivedLocalTime": time.time(),
|
|
}
|
|
|
|
log("Updated host world state")
|
|
print_host_world_state(host_world_state)
|
|
|
|
|
|
def print_host_world_state(world_state: dict[str, Any]) -> None:
|
|
client_time = world_state["clientTime"] if world_state["clientTime"] is not None else "missing from packet"
|
|
server_time = world_state["serverTime"] if world_state["serverTime"] is not None else "missing from packet"
|
|
weather_form_id = world_state["weatherFormId"] or "<none>"
|
|
|
|
log(
|
|
"\n".join(
|
|
[
|
|
f"Game Hour: {world_state['gameHour']:.3f}",
|
|
f"Game Days Passed: {world_state['gameDaysPassed']:.3f}",
|
|
f"Weather Form ID: {weather_form_id}",
|
|
f"Time Sync: {world_state.get('timeSync', False)}",
|
|
f"Client Time: {client_time}",
|
|
f"Last Server Time: {server_time}",
|
|
f"Last Received Local Time: {world_state['lastReceivedLocalTime']}",
|
|
]
|
|
)
|
|
)
|
|
|
|
|
|
def update_world_state_host(packet: dict[str, Any]) -> None:
|
|
global world_state_host_player_id
|
|
|
|
host_id = packet.get("worldStateHostPlayerId")
|
|
if host_id is None:
|
|
warn(f"Ignoring worldStateHost packet without worldStateHostPlayerId: {packet}")
|
|
return
|
|
|
|
try:
|
|
world_state_host_player_id = int(host_id)
|
|
except (TypeError, ValueError):
|
|
warn(f"Ignoring worldStateHost packet with invalid worldStateHostPlayerId: {packet}")
|
|
return
|
|
|
|
log(f"World-state host reassigned to player {world_state_host_player_id}")
|
|
|
|
|
|
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"),
|
|
"equippedItems": get_optional_equipped_items(packet),
|
|
"appearance": get_optional_appearance(packet),
|
|
"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"
|
|
equipped_items = player.get("equippedItems", [])
|
|
equipment_text = (
|
|
", ".join(f"{item['slot']}={item['formId'] or '<empty>'}" for item in equipped_items)
|
|
if equipped_items
|
|
else "<not sent>"
|
|
)
|
|
appearance = player.get("appearance")
|
|
appearance_text = (
|
|
(
|
|
f"version={appearance.get('version')}, race={appearance.get('raceFormId') or '<empty>'}, "
|
|
f"height={appearance.get('height')}, headParts={len(appearance.get('headParts', []))}, "
|
|
f"morphs={len(appearance.get('morphs', []))}, "
|
|
f"morphRegions={len(appearance.get('morphRegions', []))}, "
|
|
f"facialBoneMorphs={len(appearance.get('facialBoneMorphs', []))}, "
|
|
f"tints={len(appearance.get('tints', []))}"
|
|
)
|
|
if isinstance(appearance, dict)
|
|
else "<not sent>"
|
|
)
|
|
|
|
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"Equipment: {equipment_text}",
|
|
f"Appearance: {appearance_text}",
|
|
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]
|
|
equipment_text = (
|
|
", ".join(f"{item['slot']}={item['formId'] or '<empty>'}" for item in player.get("equippedItems", []))
|
|
if player.get("equippedItems")
|
|
else "<not sent>"
|
|
)
|
|
appearance = player.get("appearance")
|
|
appearance_text = (
|
|
f"race={appearance.get('raceFormId') or '<empty>'}, headParts={len(appearance.get('headParts', []))}, morphs={len(appearance.get('morphs', []))}, morphRegions={len(appearance.get('morphRegions', []))}, facialBoneMorphs={len(appearance.get('facialBoneMorphs', []))}, tints={len(appearance.get('tints', []))}"
|
|
if isinstance(appearance, dict)
|
|
else "<not sent>"
|
|
)
|
|
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"equipment={equipment_text}, "
|
|
f"appearance={appearance_text}, "
|
|
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":
|
|
host_id = packet.get("worldStateHostPlayerId")
|
|
if host_id is not None:
|
|
try:
|
|
global world_state_host_player_id
|
|
world_state_host_player_id = int(host_id)
|
|
log(f"Welcome packet: playerId={packet.get('playerId')}, worldStateHostPlayerId={world_state_host_player_id}")
|
|
except (TypeError, ValueError):
|
|
log(f"Welcome packet: {packet}")
|
|
else:
|
|
log(f"Welcome packet: {packet}")
|
|
elif packet_type == "transform":
|
|
log(f"Received transform: {packet}")
|
|
update_remote_player_state(packet)
|
|
elif packet_type == "worldState":
|
|
log(f"Received worldState: {packet}")
|
|
update_host_world_state(packet)
|
|
elif packet_type == "serverWorldState":
|
|
log(f"Received serverWorldState: {packet}")
|
|
elif packet_type == "worldStateHost":
|
|
log(f"Received worldStateHost: {packet}")
|
|
update_world_state_host(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}")
|