618 lines
22 KiB
Python
618 lines
22 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
|
|
received_combat_hits: list[dict[str, Any]] = []
|
|
remote_npcs: 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 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_action_events(packet: dict[str, Any]) -> list[dict[str, Any]]:
|
|
value = packet.get("actionEvents", [])
|
|
if not isinstance(value, list):
|
|
return []
|
|
|
|
action_events: list[dict[str, Any]] = []
|
|
for event in value:
|
|
if not isinstance(event, dict):
|
|
continue
|
|
|
|
event_name = event.get("eventName", "")
|
|
if not isinstance(event_name, str) or not event_name:
|
|
continue
|
|
|
|
action_events.append(
|
|
{
|
|
"sequence": get_optional_uint32(event, "sequence"),
|
|
"type": get_optional_uint32(event, "type"),
|
|
"eventName": event_name,
|
|
}
|
|
)
|
|
|
|
return action_events
|
|
|
|
|
|
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": [],
|
|
}
|
|
|
|
if isinstance(value.get("isFemale"), bool):
|
|
appearance["isFemale"] = value["isFemale"]
|
|
|
|
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
|
|
|
|
existing = remote_players.get(player_id)
|
|
character_name = packet.get("characterName", "")
|
|
if not isinstance(character_name, str):
|
|
character_name = ""
|
|
if not character_name and existing and isinstance(existing.get("characterName"), str):
|
|
character_name = existing["characterName"]
|
|
|
|
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", ""),
|
|
"characterName": character_name,
|
|
"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),
|
|
"actionEvents": get_optional_action_events(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"isFemale={appearance.get('isFemale', '<omitted>')}, "
|
|
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>"
|
|
)
|
|
action_events = player.get("actionEvents", [])
|
|
action_events_text = (
|
|
", ".join(
|
|
f"seq={event['sequence']}, type={event['type']}, event={event['eventName']}"
|
|
for event in action_events
|
|
)
|
|
if action_events
|
|
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']}",
|
|
f"Character Name: {player.get('characterName') or '<not sent>'}",
|
|
(
|
|
"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"Action Events: {action_events_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>'}, isFemale={appearance.get('isFemale', '<omitted>')}, 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_session_ended_packet(packet: dict[str, Any]) -> None:
|
|
code = packet.get("code", "")
|
|
reason = packet.get("reason", "")
|
|
reason_text = f", reason={reason!r}" if reason else ""
|
|
log(f"Received sessionEnded: code={code!r}{reason_text}")
|
|
|
|
|
|
def handle_combat_hit_packet(packet: dict[str, Any]) -> None:
|
|
sender_id = get_optional_uint32(packet, "playerId")
|
|
target_player_id = get_optional_uint32(packet, "targetPlayerId")
|
|
sequence = get_optional_uint32(packet, "sequence")
|
|
damage = get_optional_float(packet, "damage", -1.0)
|
|
if sender_id == 0 or target_player_id == 0 or sequence == 0 or damage <= 0.0:
|
|
warn(f"Ignoring malformed combatHit packet: {packet}")
|
|
return
|
|
|
|
received_combat_hits.append(dict(packet))
|
|
log(
|
|
f"Received combatHit: sender={sender_id}, target={target_player_id}, "
|
|
f"sequence={sequence}, damage={damage:.1f}"
|
|
)
|
|
|
|
|
|
def update_remote_npc_state(packet: dict[str, Any]) -> None:
|
|
npcs = packet.get("npcs")
|
|
if not isinstance(npcs, list):
|
|
warn(f"Ignoring npcState packet without an npcs array: {packet}")
|
|
return
|
|
|
|
next_snapshot: dict[int, dict[str, Any]] = {}
|
|
for npc in npcs[:16]:
|
|
if not isinstance(npc, dict):
|
|
continue
|
|
npc_id = get_optional_uint32(npc, "npcId")
|
|
base_form_id = npc.get("baseFormId")
|
|
if npc_id == 0 or not isinstance(base_form_id, str):
|
|
continue
|
|
next_snapshot[npc_id] = dict(npc)
|
|
|
|
remote_npcs.clear()
|
|
remote_npcs.update(next_snapshot)
|
|
log(
|
|
f"Updated host NPC snapshot: count={len(remote_npcs)}, "
|
|
f"npcIds={sorted(remote_npcs)}"
|
|
)
|
|
|
|
|
|
def handle_packet(packet: dict[str, Any]) -> None:
|
|
packet_type = packet.get("type")
|
|
if packet_type == "welcome":
|
|
host_id = packet.get("worldStateHostPlayerId")
|
|
server_name = packet.get("serverName")
|
|
server_description = packet.get("serverDescription")
|
|
meta_parts = [f"playerId={packet.get('playerId')}"]
|
|
if server_name is not None:
|
|
meta_parts.append(f"serverName={server_name!r}")
|
|
if server_description is not None:
|
|
meta_parts.append(f"serverDescription={server_description!r}")
|
|
if host_id is not None:
|
|
try:
|
|
global world_state_host_player_id
|
|
world_state_host_player_id = int(host_id)
|
|
meta_parts.append(f"worldStateHostPlayerId={world_state_host_player_id}")
|
|
log(f"Welcome packet: {', '.join(meta_parts)}")
|
|
except (TypeError, ValueError):
|
|
log(f"Welcome packet: {packet}")
|
|
else:
|
|
log(f"Welcome packet: {', '.join(meta_parts)}")
|
|
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)
|
|
elif packet_type == "sessionEnded":
|
|
handle_session_ended_packet(packet)
|
|
elif packet_type == "combatHit":
|
|
handle_combat_hit_packet(packet)
|
|
elif packet_type == "npcState":
|
|
update_remote_npc_state(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 send_ranged_fire_action(socket_connection: socket.socket, sequence: int) -> None:
|
|
"""Send a simulated ranged fire action event (type=3, eventName=fireSingle)."""
|
|
transform_packet = {
|
|
"type": "transform",
|
|
"x": -1792.0,
|
|
"y": -1344.0,
|
|
"z": 0.0,
|
|
"angleZ": 3.93,
|
|
"movementType": "normal",
|
|
"isMoving": False,
|
|
"isSprinting": False,
|
|
"isSneaking": False,
|
|
"isJumping": False,
|
|
"isCrouching": False,
|
|
"weaponDrawn": True,
|
|
"movementSpeed": 0.0,
|
|
"actionEvents": [
|
|
{
|
|
"sequence": sequence,
|
|
"type": 3,
|
|
"eventName": "fireSingle",
|
|
}
|
|
],
|
|
}
|
|
packet_line = json.dumps(transform_packet, separators=(",", ":"))
|
|
socket_connection.sendall((packet_line + "\n").encode("utf-8"))
|
|
log(f"Sent ranged fire action: sequence={sequence}, type=3, eventName=fireSingle")
|
|
|
|
|
|
def send_combat_hit(socket_connection: socket.socket, target_player_id: int, sequence: int, damage: float, weapon_form_id: int = 0) -> None:
|
|
"""Send a targeted combatHit packet."""
|
|
packet = {
|
|
"type": "combatHit",
|
|
"sequence": sequence,
|
|
"targetPlayerId": target_player_id,
|
|
"damage": damage,
|
|
}
|
|
if weapon_form_id != 0:
|
|
packet["weaponFormId"] = f"{weapon_form_id:08X}"
|
|
|
|
packet_line = json.dumps(packet, separators=(",", ":"))
|
|
socket_connection.sendall((packet_line + "\n").encode("utf-8"))
|
|
log(f"Sent combatHit: targetPlayerId={target_player_id}, damage={damage:.1f}, weaponFormId={weapon_form_id:08X}")
|
|
|
|
|
|
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}")
|