Skip to content

Instantly share code, notes, and snippets.

@jun-lsh

jun-lsh/solve.py

Created May 31, 2026 12:53
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save jun-lsh/ecd7d231182d1559627a6c21bb593a4b to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save jun-lsh/ecd7d231182d1559627a6c21bb593a4b to your computer and use it in GitHub Desktop.
Download ZIP
2bird2can solve script
Raw
solve.py
#!/usr/bin/env python3
import argparse
import concurrent.futures
import functools
import hashlib
import math
import os
import re
import struct
import subprocess
import sys
import time
import urllib.request
import zlib
import websocket
import xxhash
SIG = 0x01505455
PLAYER_HASH = 0x3D2E5A4B
FLAG_CHEST_HASH = 0x5D124AEB
GAME_MANAGER_HASH = 0x7866F579
CONFIG_HASH = 0x21DCDE43F7A3E2F0
MOVE_RPC = 0x6CBBEDD4
FLAG_RPC = 0x0729BFCF
CHUNK_RPC = 0x3A8A70E0
MASK48 = (1 << 48) - 1
SEED_LOW_MASK = (1 << 18) - 1
WORLD_RAND_MULT = 0x5DEECE66D
WORLD_RAND_ADD = 0xB
MIX_X = 0x4F9939F508
MIX_Y = 0x1EF1565BD5
MESSAGE_NAMES = [
"ConnectionApprovedMessage",
"ConnectionRequestMessage",
"ChangeOwnershipMessage",
"ClientConnectedMessage",
"ClientDisconnectedMessage",
"ClientRpcMessage",
"CreateObjectMessage",
"DestroyObjectMessage",
"DisconnectReasonMessage",
"ForwardClientRpcMessage",
"ForwardServerRpcMessage",
"NamedMessage",
"NetworkTransformMessage",
"NetworkVariableDeltaMessage",
"ParentSyncMessage",
"ProxyMessage",
"RpcMessage",
"SceneEventMessage",
"ServerLogMessage",
"ServerRpcMessage",
"TimeSyncMessage",
"UnnamedMessage",
"SessionOwnerMessage",
"AnticipationCounterSyncPingMessage",
"AnticipationCounterSyncPongMessage",
]
MESSAGE_VERSIONS = {
"ConnectionApprovedMessage": 2,
"ConnectionRequestMessage": 1,
"DestroyObjectMessage": 2,
"NetworkVariableDeltaMessage": 1,
}
def bitpack_uint(value, bits=32):
if bits == 16:
if value > (1 << 14) - 1:
return bytes([3]) + struct.pack("<H", value)
shifted = value << 2
elif bits == 32:
if value > (1 << 29) - 1:
return bytes([5]) + struct.pack("<I", value)
shifted = value << 3
elif bits == 64:
if value > (1 << 60) - 1:
return bytes([9]) + struct.pack("<Q", value)
shifted = value << 4
else:
raise ValueError(bits)
nbytes = max(1, (shifted.bit_length() + 7) // 8)
return int(shifted | nbytes).to_bytes(nbytes, "little")
def bitpack_int(value):
zigzag = ((value << 1) ^ (value >> 31)) & 0xFFFFFFFF
return bitpack_uint(zigzag, 32)
def read_bit_uint(data, pos, bits=32):
mask = {16: 3, 32: 7, 64: 15}[bits]
nbytes = data[pos] & mask
if bits == 16:
if nbytes == 3:
return struct.unpack_from("<H", data, pos + 1)[0], pos + 3
return int.from_bytes(data[pos : pos + nbytes], "little") >> 2, pos + nbytes
if bits == 32:
if nbytes == 5:
return struct.unpack_from("<I", data, pos + 1)[0], pos + 5
return int.from_bytes(data[pos : pos + nbytes], "little") >> 3, pos + nbytes
if nbytes == 9:
return struct.unpack_from("<Q", data, pos + 1)[0], pos + 9
return int.from_bytes(data[pos : pos + nbytes], "little") >> 4, pos + nbytes
def read_bit_int(data, pos):
raw, pos = read_bit_uint(data, pos, 32)
return ((raw >> 1) ^ -(raw & 1)), pos
def make_batch(message_type, body):
message = bitpack_uint(message_type, 32) + bitpack_uint(len(body), 32) + body
aligned = (16 + len(message) + 7) & ~7
payload = message + b"\x00" * (aligned - 16 - len(message))
return struct.pack("<HHIQ", 0x1160, 1, aligned, xxhash.xxh64(payload).intdigest()) + payload
def write_string(value):
return struct.pack("<I", len(value)) + value.encode("utf-16le")
def connection_request_body():
body = bitpack_int(len(MESSAGE_NAMES))
for name in MESSAGE_NAMES:
full_name = "Unity.Netcode." + name
body += struct.pack("<I", xxhash.xxh32(full_name.encode()).intdigest())
body += bitpack_int(MESSAGE_VERSIONS.get(name, 0))
body += struct.pack("<Q", CONFIG_HASH)
body += struct.pack("<I", 0)
return body
def server_rpc_body(network_object_id, behaviour_id, method_id, payload):
return (
bitpack_uint(network_object_id, 64)
+ bitpack_uint(behaviour_id, 16)
+ bitpack_uint(method_id, 32)
+ payload
)
def scene_sync_complete_body(scene_hash, scene_handle, object_ids):
body = bytearray()
body += b"\x08" # SceneEventType.SynchronizeComplete
body += b"\x00" # LoadSceneMode.Single
body += b"\x00" * 16 # default Guid; not used for synchronize-complete server handling
body += struct.pack("<I", scene_hash)
body += struct.pack("<i", scene_handle)
body += struct.pack("<I", len(object_ids))
for object_id in object_ids:
body += struct.pack("<I", object_id & 0xFFFFFFFF)
return bytes(body)
def reliable_ack_frame(token, sequence_id):
# ReliableUtility.PacketHeader + 8-byte ack mask + dummy payload byte.
header = struct.pack("<BBHHH", 1, 0, 0, 0, sequence_id & 0xFFFF)
return b"\x01" + token + b"\x03" + header + (b"\xff" * 8) + b"\x00"
def simple_data_frame(token, pipeline_id, payload):
return b"\x01" + token + bytes([pipeline_id]) + payload
def batch_frame(token, message_type, body):
batch = make_batch(message_type, body)
return simple_data_frame(token, 0, struct.pack("<i", len(batch)) + batch)
def extract_batches(packet, token):
if not isinstance(packet, bytes) or len(packet) < 10:
return []
if packet[0] != 1 or packet[1:9] != token:
return []
pipeline_id = packet[9]
payload = packet[10:]
batches = []
reliable_sequence = None
if pipeline_id == 3 and len(payload) >= 16:
packet_type, _, _, reliable_sequence, _ = struct.unpack_from("<BBHHH", payload, 0)
if packet_type == 1:
return []
payload = payload[16:]
elif pipeline_id == 1 and len(payload) >= 2:
payload = payload[2:]
start = None
for off in range(min(40, max(0, len(payload) - 6))):
size = struct.unpack_from("<i", payload, off)[0]
if 16 <= size <= 100000 and off + 4 + size <= len(payload):
if payload[off + 4 : off + 6] == b"\x60\x11":
start = off
break
if start is None:
return []
pos = start
while pos + 4 <= len(payload):
size = struct.unpack_from("<i", payload, pos)[0]
if not (16 <= size <= 100000 and pos + 4 + size <= len(payload)):
break
batch = payload[pos + 4 : pos + 4 + size]
if batch[:2] != b"\x60\x11":
break
batches.append((pipeline_id, reliable_sequence, batch))
pos += 4 + size
return batches
def parse_messages(batch):
count = struct.unpack_from("<H", batch, 2)[0]
pos = 16
for _ in range(count):
message_type, pos = read_bit_uint(batch, pos, 32)
message_size, pos = read_bit_uint(batch, pos, 32)
body = batch[pos : pos + message_size]
pos += message_size
yield message_type, body
def parse_connection_approved(body):
try:
pos = 0
message_count, pos = read_bit_int(body, pos)
for _ in range(message_count):
pos += 4
_, pos = read_bit_int(body, pos)
owner_client_id, pos = read_bit_uint(body, pos, 64)
return owner_client_id
except (IndexError, struct.error):
return None
def parse_scene_sync(body, local_client_id=None):
pos = 0
event_type = body[pos]
pos += 1
if event_type != 2:
return None
pos += 1 # LoadSceneMode byte
pos += 4 # ClientSynchronizationMode
scene_hash = struct.unpack_from("<I", body, pos)[0]
pos += 4
scene_handle = struct.unpack_from("<i", body, pos)[0]
pos += 4
pos += 4 # active scene hash
scene_count = struct.unpack_from("<i", body, pos)[0]
pos += 4 + (4 * scene_count)
handle_count = struct.unpack_from("<i", body, pos)[0]
pos += 4 + (4 * handle_count)
sync_size = struct.unpack_from("<i", body, pos)[0]
pos += 4
sync_end = pos + sync_size
object_count = struct.unpack_from("<i", body, pos)[0]
pos += 4
object_ids = []
player_candidates = []
game_manager_object_id = None
for _ in range(object_count):
flags = struct.unpack_from("<H", body, pos)[0]
pos += 2
object_hash = struct.unpack_from("<I", body, pos)[0]
pos += 4
object_id, pos = read_bit_uint(body, pos, 64)
owner_id, pos = read_bit_uint(body, pos, 64)
object_ids.append(object_id)
if flags & 0x002:
_, pos = read_bit_uint(body, pos, 64)
if flags & 0x010:
_, pos = read_bit_uint(body, pos, 64)
if flags & 0x100:
pos += 2
if flags & 0x200:
observer_count, pos = read_bit_uint(body, pos, 32)
for _ in range(observer_count):
_, pos = read_bit_uint(body, pos, 64)
if flags & 0x008:
pos += 40
pos += 4
sync_data_size = struct.unpack_from("<i", body, pos)[0]
pos += 4 + sync_data_size
if object_hash == PLAYER_HASH:
player_candidates.append((object_id, owner_id))
elif object_hash == GAME_MANAGER_HASH:
game_manager_object_id = object_id
pos += 4 # despawned in-scene object count
if pos != sync_end:
pass
player_object_id = None
owner_client_id = None
if local_client_id is not None:
for object_id, owner_id in player_candidates:
if owner_id == local_client_id:
player_object_id = object_id
owner_client_id = owner_id
break
if player_object_id is None and player_candidates:
player_object_id, owner_client_id = player_candidates[-1]
return scene_hash, scene_handle, object_ids, player_object_id, owner_client_id, game_manager_object_id
def parse_client_rpc_header(body):
pos = 0
object_id, pos = read_bit_uint(body, pos, 64)
behaviour_id, pos = read_bit_uint(body, pos, 16)
method_id, pos = read_bit_uint(body, pos, 32)
return object_id, behaviour_id, method_id, body[pos:]
def parse_flag_client_rpc_payload(payload):
marker = payload.find(b"bbb{")
if marker >= 0:
end = payload.find(b"}", marker)
if end >= 0:
return payload[marker : end + 1].decode("ascii", errors="ignore")
utf16_marker = payload.find("bbb{".encode("utf-16le"))
if utf16_marker >= 0:
utf16_end = payload.find("}".encode("utf-16le"), utf16_marker)
if utf16_end >= 0:
raw = payload[utf16_marker : utf16_end + 2]
try:
return raw.decode("utf-16le")
except UnicodeDecodeError:
pass
if len(payload) < 4:
return None
length = struct.unpack_from("<I", payload, 0)[0]
raw = payload[4 : 4 + length * 2]
try:
decoded = raw.decode("utf-16le")
return decoded if decoded else None
except UnicodeDecodeError:
return None
def parse_client_rpc(body):
pos = 0
object_id, pos = read_bit_uint(body, pos, 64)
behaviour_id, pos = read_bit_uint(body, pos, 16)
method_id, pos = read_bit_uint(body, pos, 32)
payload = body[pos:]
if method_id != FLAG_RPC:
return None
return parse_flag_client_rpc_payload(payload)
def parse_chunk_rpc_payload(payload):
try:
pos = 0
chunk_x, pos = read_bit_int(payload, pos)
chunk_y, pos = read_bit_int(payload, pos)
decoration_count = payload[pos]
pos += 1
decoration_xs = list(struct.unpack_from("<" + "f" * decoration_count, payload, pos))
pos += 4 * decoration_count
decoration_ys = list(struct.unpack_from("<" + "f" * decoration_count, payload, pos))
pos += 4 * decoration_count
decoration_variants = list(payload[pos : pos + decoration_count])
pos += decoration_count
has_structure = bool(payload[pos])
pos += 1
structure_type = payload[pos] if pos < len(payload) else 0
return {
"chunk_x": chunk_x,
"chunk_y": chunk_y,
"decoration_xs": decoration_xs,
"decoration_ys": decoration_ys,
"decoration_variants": decoration_variants,
"has_structure": has_structure,
"structure_type": structure_type,
}
except (IndexError, struct.error):
return None
def parse_network_variable_delta(body):
try:
pos = 0
object_id, pos = read_bit_uint(body, pos, 64)
behaviour_id, pos = read_bit_uint(body, pos, 16)
delivery = struct.unpack_from("<i", body, pos)[0]
pos += 4
mask = body[pos]
pos += 1
if mask & 1 and pos + 8 <= len(body):
x, y = struct.unpack_from("<ff", body, pos)
return object_id, behaviour_id, delivery, mask, x, y
except (IndexError, struct.error):
return None
return None
class NgoClient:
def __init__(self, url, timeout=1.0):
self.url = url
self.timeout = timeout
last_error = None
for _ in range(30):
try:
self.ws = websocket.create_connection(url, timeout=timeout)
break
except Exception as exc:
last_error = exc
time.sleep(0.25)
else:
raise last_error
self.token = os.urandom(8)
self.player_object_id = None
self.owner_client_id = None
self.scene_hash = None
self.scene_handle = None
self.synced_object_ids = []
self.game_manager_object_id = 1
self.position = (0.0, 0.0)
self.wind = (0.0, 0.0)
self.flag = None
self.chunks = {}
def close(self):
try:
self.ws.close()
except Exception:
pass
def send_connection_request(self):
self.ws.send_binary(struct.pack("<I", SIG) + b"\x01" + self.token)
self.ws.recv()
self.ws.send_binary(batch_frame(self.token, 1, connection_request_body()))
def read_messages(self, duration=0.5):
deadline = time.time() + duration
while time.time() < deadline:
try:
self.ws.settimeout(max(0.05, min(self.timeout, deadline - time.time())))
packet = self.ws.recv()
except Exception:
continue
for _, sequence_id, batch in extract_batches(packet, self.token):
if sequence_id is not None:
try:
self.ws.send_binary(reliable_ack_frame(self.token, sequence_id))
except Exception:
return
for message_type, body in parse_messages(batch):
yield message_type, body
def connect_and_sync(self):
self.send_connection_request()
deadline = time.time() + 5.0
while time.time() < deadline and self.player_object_id is None:
for message_type, body in self.read_messages(0.5):
if message_type == 0:
approved_id = parse_connection_approved(body)
if approved_id is not None:
self.owner_client_id = approved_id
elif message_type == 17:
parsed = parse_scene_sync(body, self.owner_client_id)
if parsed:
(
self.scene_hash,
self.scene_handle,
self.synced_object_ids,
self.player_object_id,
self.owner_client_id,
self.game_manager_object_id,
) = parsed
break
elif message_type == 8:
raise RuntimeError("server disconnected during approval")
if self.player_object_id is None:
raise RuntimeError("did not receive scene synchronization")
body = scene_sync_complete_body(self.scene_hash, self.scene_handle, self.synced_object_ids)
self.ws.send_binary(batch_frame(self.token, 17, body))
# Give the server a moment to mark the client connected.
self.pump_state(0.25)
def connect_quick_sync(self):
self.send_connection_request()
deadline = time.time() + 3.0
while time.time() < deadline and self.player_object_id is None:
for message_type, body in self.read_messages(0.2):
if message_type == 0:
approved_id = parse_connection_approved(body)
if approved_id is not None:
self.owner_client_id = approved_id
elif message_type == 17:
parsed = parse_scene_sync(body, self.owner_client_id)
if parsed:
(
self.scene_hash,
self.scene_handle,
self.synced_object_ids,
self.player_object_id,
self.owner_client_id,
self.game_manager_object_id,
) = parsed
break
elif message_type == 8:
raise RuntimeError("server disconnected helper during approval")
if self.player_object_id is None:
raise RuntimeError("helper did not receive scene synchronization")
body = scene_sync_complete_body(self.scene_hash, self.scene_handle, self.synced_object_ids)
self.ws.send_binary(batch_frame(self.token, 17, body))
self.pump_state(0.08)
def connect_fast_sync(self):
self.send_connection_request()
# Helpers do not need to instantiate the scene locally. A synchronize-complete
# event with an empty object list is enough for the server-side connection path.
list(self.read_messages(0.15))
body = scene_sync_complete_body(0, 0, [])
self.ws.send_binary(batch_frame(self.token, 17, body))
list(self.read_messages(0.10))
def send_move(self, target_object_id, x, y, sender_id=None):
payload = struct.pack("<ff", float(x), float(y))
body = bitpack_uint(sender_id if sender_id is not None else (self.owner_client_id or 0), 64)
body += server_rpc_body(target_object_id, 0, MOVE_RPC, payload)
self.ws.send_binary(batch_frame(self.token, 16, body))
def pump_state(self, duration=0.1):
for message_type, body in self.read_messages(duration):
if message_type == 13:
delta = parse_network_variable_delta(body)
if not delta:
continue
object_id, _, _, _, x, y = delta
if object_id == self.player_object_id:
self.position = (x, y)
elif object_id == self.game_manager_object_id:
self.wind = (x, y)
elif message_type == 5:
try:
_, _, method_id, payload = parse_client_rpc_header(body)
except (IndexError, struct.error):
continue
if method_id == FLAG_RPC:
flag = parse_flag_client_rpc_payload(payload)
if flag:
self.flag = flag
return flag
elif method_id == CHUNK_RPC:
chunk = parse_chunk_rpc_payload(payload)
if chunk:
self.chunks[(chunk["chunk_x"], chunk["chunk_y"])] = chunk
return None
def poll_flag(self, duration=0.05):
for message_type, body in self.read_messages(duration):
if message_type == 5:
try:
_, _, method_id, payload = parse_client_rpc_header(body)
except (IndexError, struct.error):
continue
if method_id == FLAG_RPC:
flag = parse_flag_client_rpc_payload(payload)
if flag:
self.flag = flag
return flag
elif method_id == CHUNK_RPC:
chunk = parse_chunk_rpc_payload(payload)
if chunk:
self.chunks[(chunk["chunk_x"], chunk["chunk_y"])] = chunk
return None
def read_current_docker_logs(wait_seconds=5.0):
since_args = []
try:
container_id = subprocess.check_output(
["docker", "compose", "ps", "-q", "server"],
text=True,
stderr=subprocess.DEVNULL,
).strip()
if container_id:
started_at = subprocess.check_output(
["docker", "inspect", "-f", "{{.State.StartedAt}}", container_id],
text=True,
stderr=subprocess.DEVNULL,
).strip()
if started_at:
since_args = ["--since", started_at]
except Exception:
since_args = []
deadline = time.time() + wait_seconds
last_output = ""
while True:
try:
output = subprocess.check_output(
["docker", "compose", "logs", *since_args, "--tail=2000", "server"],
text=True,
stderr=subprocess.DEVNULL,
)
if output:
return output
last_output = output
except Exception as exc:
last_output = str(exc)
if time.time() >= deadline:
return last_output
time.sleep(0.25)
def read_target_from_docker_logs():
try:
deadline = time.time() + 7.0
output = ""
while time.time() < deadline:
output = read_current_docker_logs(wait_seconds=0.1)
matches = re.findall(
r"flag chest prepared at \(([-0-9.]+),\s*([-0-9.]+),\s*[-0-9.]+\)",
output,
)
if matches:
x, y = matches[-1]
return float(x), float(y)
time.sleep(0.25)
except Exception as exc:
raise RuntimeError(f"could not read docker logs: {exc}") from exc
matches = re.findall(
r"flag chest prepared at \(([-0-9.]+),\s*([-0-9.]+),\s*[-0-9.]+\)",
output,
)
if not matches:
output = subprocess.check_output(
["docker", "compose", "logs", "server"],
text=True,
stderr=subprocess.DEVNULL,
)
matches = re.findall(
r"flag chest prepared at \(([-0-9.]+),\s*([-0-9.]+),\s*[-0-9.]+\)",
output,
)
if not matches:
raise RuntimeError("could not find flag coordinates in docker logs")
x, y = matches[-1]
return float(x), float(y)
def read_seed_from_docker_logs():
try:
deadline = time.time() + 7.0
output = ""
while time.time() < deadline:
output = read_current_docker_logs(wait_seconds=0.1)
matches = re.findall(r"seed:\s*([0-9]+)", output)
if matches:
return int(matches[-1])
time.sleep(0.25)
except Exception as exc:
raise RuntimeError(f"could not read docker logs: {exc}") from exc
matches = re.findall(r"seed:\s*([0-9]+)", output)
if not matches:
output = subprocess.check_output(
["docker", "compose", "logs", "server"],
text=True,
stderr=subprocess.DEVNULL,
)
matches = re.findall(r"seed:\s*([0-9]+)", output)
if not matches:
raise RuntimeError("could not find world seed in docker logs")
return int(matches[-1])
def unit_towards(pos_x, pos_y, target_x, target_y):
dx = target_x - pos_x
dy = target_y - pos_y
dist = math.hypot(dx, dy)
if dist < 1e-6:
return 0.0, 0.0, dist
return dx / dist, dy / dist, dist
def vector_len(x, y):
return math.hypot(x, y)
def normalize(x, y):
length = vector_len(x, y)
if length < 1e-6:
return 0.0, 0.0
return x / length, y / length
def world_to_chunk(x, y):
return math.floor(x / 16.0), math.floor(y / 16.0)
def chunk_to_region(chunk_x, chunk_y):
return chunk_x // 4, chunk_y // 4
def structure_chunk(seed, region_x, region_y):
state = (seed ^ (region_x * MIX_X) ^ (region_y * MIX_Y)) & MASK48
state = (state * WORLD_RAND_MULT + WORLD_RAND_ADD) & MASK48
dx = ((state >> 16) & 0x7FFFFFFF) % 4
state = (state * WORLD_RAND_MULT + WORLD_RAND_ADD) & MASK48
dy = ((state >> 16) & 0x7FFFFFFF) % 4
return region_x * 4 + dx, region_y * 4 + dy
def chunk_has_structure_for_seed(seed, chunk_x, chunk_y):
region_x, region_y = chunk_to_region(chunk_x, chunk_y)
return structure_chunk(seed, region_x, region_y) == (chunk_x, chunk_y)
def structure_constraints_from_chunks(chunks):
constraints = {}
for (chunk_x, chunk_y), chunk in chunks.items():
region = chunk_to_region(chunk_x, chunk_y)
required, forbidden = constraints.setdefault(region, [None, set()])
pos = (chunk_x, chunk_y)
if chunk["has_structure"]:
if required is not None and required != pos:
raise RuntimeError(f"conflicting structure observations in region {region}")
constraints[region][0] = pos
else:
forbidden.add(pos)
return constraints
def seed_low_candidates_from_chunks(chunks):
constraints = structure_constraints_from_chunks(chunks)
candidates = []
for seed_low in range(SEED_LOW_MASK + 1):
ok = True
for (region_x, region_y), (required, forbidden) in constraints.items():
structure = structure_chunk(seed_low, region_x, region_y)
if required is not None:
if structure != required:
ok = False
break
elif structure in forbidden:
ok = False
break
if ok:
candidates.append(seed_low)
return candidates
def fetch_url(url, timeout=5.0):
last_error = None
for _ in range(30):
try:
with urllib.request.urlopen(url, timeout=timeout) as response:
return response.read()
except Exception as exc:
last_error = exc
time.sleep(0.25)
raise last_error
def paeth_predictor(a, b, c):
p = a + b - c
pa = abs(p - a)
pb = abs(p - b)
pc = abs(p - c)
if pa <= pb and pa <= pc:
return a
if pb <= pc:
return b
return c
def decode_png_rgba(data):
if data[:8] != b"\x89PNG\r\n\x1a\n":
raise RuntimeError("map response is not a PNG")
pos = 8
width = height = bit_depth = color_type = interlace = None
compressed = bytearray()
while pos + 8 <= len(data):
length = struct.unpack_from(">I", data, pos)[0]
chunk_type = data[pos + 4 : pos + 8]
chunk_data = data[pos + 8 : pos + 8 + length]
pos += 12 + length
if chunk_type == b"IHDR":
width, height, bit_depth, color_type, _, _, interlace = struct.unpack(">IIBBBBB", chunk_data)
elif chunk_type == b"IDAT":
compressed += chunk_data
elif chunk_type == b"IEND":
break
if width is None or height is None:
raise RuntimeError("PNG is missing IHDR")
if bit_depth != 8 or color_type != 6 or interlace != 0:
raise RuntimeError("expected an 8-bit non-interlaced RGBA PNG")
raw = zlib.decompress(bytes(compressed))
stride = width * 4
rows = []
prev = bytearray(stride)
pos = 0
for _ in range(height):
filter_type = raw[pos]
pos += 1
row = bytearray(raw[pos : pos + stride])
pos += stride
for i, value in enumerate(row):
left = row[i - 4] if i >= 4 else 0
up = prev[i]
up_left = prev[i - 4] if i >= 4 else 0
if filter_type == 1:
row[i] = (value + left) & 0xFF
elif filter_type == 2:
row[i] = (value + up) & 0xFF
elif filter_type == 3:
row[i] = (value + ((left + up) // 2)) & 0xFF
elif filter_type == 4:
row[i] = (value + paeth_predictor(left, up, up_left)) & 0xFF
elif filter_type != 0:
raise RuntimeError(f"unsupported PNG filter {filter_type}")
rows.append(bytes(row))
prev = row
return width, height, rows
MAP_STRUCTURE_COLORS = {
(232, 212, 150, 255), # beach
(210, 180, 110, 255), # island
(120, 40, 30, 255), # flag X
}
MAP_FLAG_COLOR = (120, 40, 30, 255)
def parse_map_structure_offsets(png_data):
width, height, rows = decode_png_rgba(png_data)
if width != height or width % 16 != 0:
raise RuntimeError(f"unexpected map dimensions {width}x{height}")
cell = width // 16
structure_cells = set()
red_counts = {}
for cell_y in range(16):
for cell_x in range(16):
has_structure = False
red_count = 0
for y in range(cell_y * cell, (cell_y + 1) * cell):
row = rows[y]
for x in range(cell_x * cell, (cell_x + 1) * cell):
off = x * 4
color = tuple(row[off : off + 4])
if color in MAP_STRUCTURE_COLORS:
has_structure = True
if color == MAP_FLAG_COLOR:
red_count += 1
if has_structure:
structure_cells.add((cell_x, cell_y))
if red_count:
red_counts[(cell_x, cell_y)] = red_count
if not red_counts:
raise RuntimeError("map does not contain a flag X marker")
flag_cell = max(red_counts, key=red_counts.get)
flag_x, flag_y = flag_cell
offsets = {(cell_x - flag_x, flag_y - cell_y) for cell_x, cell_y in structure_cells}
if (0, 0) not in offsets:
raise RuntimeError("flag marker cell was not parsed as a structure")
bounds = (-flag_x, 15 - flag_x, flag_y - 15, flag_y)
return offsets, bounds, flag_cell
def expected_structure_offsets(seed, flag_chunk_x, flag_chunk_y, bounds):
min_dx, max_dx, min_dy, max_dy = bounds
min_x = flag_chunk_x + min_dx
max_x = flag_chunk_x + max_dx
min_y = flag_chunk_y + min_dy
max_y = flag_chunk_y + max_dy
offsets = set()
for region_y in range(min_y // 4 - 1, max_y // 4 + 2):
for region_x in range(min_x // 4 - 1, max_x // 4 + 2):
sx, sy = structure_chunk(seed, region_x, region_y)
if min_x <= sx <= max_x and min_y <= sy <= max_y:
offsets.add((sx - flag_chunk_x, sy - flag_chunk_y))
return offsets
def iter_flag_candidate_regions(min_radius=1898, max_radius=2102):
min_sq = min_radius * min_radius
max_sq = max_radius * max_radius
for region_y in range(-max_radius, max_radius + 1):
x_limit = int(math.floor(math.sqrt(max(0, max_sq - region_y * region_y))))
inner = min_sq - region_y * region_y
x_inner = math.ceil(math.sqrt(inner)) if inner > 0 else 0
if x_inner > 0:
ranges = (range(-x_limit, -x_inner + 1), range(x_inner, x_limit + 1))
else:
ranges = (range(-x_limit, x_limit + 1),)
for region_range in ranges:
for region_x in region_range:
yield region_x, region_y
def find_flag_chunk_from_map(seed, png_data):
observed_offsets, bounds, flag_cell = parse_map_structure_offsets(png_data)
filters = [offset for offset in observed_offsets if offset != (0, 0)]
filters.sort(key=lambda item: -(abs(item[0]) + abs(item[1])))
matches = []
for region_x, region_y in iter_flag_candidate_regions():
flag_chunk_x, flag_chunk_y = structure_chunk(seed, region_x, region_y)
ok = True
for dx, dy in filters[:8]:
if not chunk_has_structure_for_seed(seed, flag_chunk_x + dx, flag_chunk_y + dy):
ok = False
break
if not ok:
continue
expected = expected_structure_offsets(seed, flag_chunk_x, flag_chunk_y, bounds)
if expected == observed_offsets:
matches.append((flag_chunk_x, flag_chunk_y, region_x, region_y))
if len(matches) > 1:
break
if len(matches) != 1:
raise RuntimeError(
f"map fingerprint was not unique for seed-low {seed & SEED_LOW_MASK}: "
f"{len(matches)} matches, flag_cell={flag_cell}"
)
return matches[0]
@functools.lru_cache(maxsize=131072)
def rock_center(seed, chunk_x, chunk_y):
digest = hashlib.sha256(struct.pack("<qii", seed, chunk_x, chunk_y)).digest()
return chunk_x * 16.0 + digest[1] / 255.0 * 16.0, chunk_y * 16.0 + digest[2] / 255.0 * 16.0
def local_avoidance_direction(
seed,
pos_x,
pos_y,
target_x,
target_y,
base_ux,
base_uy,
lookahead,
include_rocks=True,
):
side_x, side_y = -base_uy, base_ux
chunk_x, chunk_y = world_to_chunk(pos_x, pos_y)
radius_chunks = int(math.ceil(lookahead / 16.0)) + 3
best = None
def consider(ox, oy, clearance, weight):
nonlocal best
rel_x = ox - pos_x
rel_y = oy - pos_y
ahead = rel_x * base_ux + rel_y * base_uy
if ahead <= 0.0 or ahead > lookahead:
return
lateral = rel_x * side_x + rel_y * side_y
margin = clearance + 1.0
if abs(lateral) > margin:
return
score = ahead - (margin - abs(lateral)) * weight
if best is None or score < best[0]:
best = (score, lateral, margin, ahead, weight)
region_x = math.floor(chunk_x / 4)
region_y = math.floor(chunk_y / 4)
region_radius = radius_chunks // 4 + 2
for ry in range(region_y - region_radius, region_y + region_radius + 1):
for rx in range(region_x - region_radius, region_x + region_radius + 1):
sx, sy = structure_chunk(seed, rx, ry)
consider(sx * 16.0 + 8.0, sy * 16.0 + 8.0, 8.5, 2.2)
if include_rocks:
for cy in range(chunk_y - radius_chunks, chunk_y + radius_chunks + 1):
for cx in range(chunk_x - radius_chunks, chunk_x + radius_chunks + 1):
ox, oy = rock_center(seed, cx, cy)
consider(ox, oy, 2.6, 0.9)
if best is None:
return base_ux, base_uy, False
_, lateral, margin, ahead, weight = best
sign = -1.0 if lateral > 0.0 else 1.0
strength = min(2.4, 0.85 + (margin - abs(lateral)) / max(margin, 1.0) * weight)
# Prefer a tangent-like heading around the first upcoming collider.
return normalize(base_ux + side_x * sign * strength, base_uy + side_y * sign * strength) + (True,)
def local_escape_direction(seed, pos_x, pos_y, include_rocks=True):
chunk_x, chunk_y = world_to_chunk(pos_x, pos_y)
best = None
def record(dx, dy, clearance, weight):
nonlocal best
dist = vector_len(dx, dy)
if dist > clearance:
return
if dist < 1e-4:
ux, uy = 1.0, 0.0
else:
ux, uy = dx / dist, dy / dist
score = (clearance - dist) * weight
if best is None or score > best[0]:
best = (score, ux, uy)
if include_rocks:
for cy in range(chunk_y - 2, chunk_y + 3):
for cx in range(chunk_x - 2, chunk_x + 3):
ox, oy = rock_center(seed, cx, cy)
record(pos_x - ox, pos_y - oy, 4.5, 1.0)
region_x = math.floor(chunk_x / 4)
region_y = math.floor(chunk_y / 4)
for ry in range(region_y - 2, region_y + 3):
for rx in range(region_x - 2, region_x + 3):
sx, sy = structure_chunk(seed, rx, ry)
cx = sx * 16.0 + 8.0
cy = sy * 16.0 + 8.0
dx = pos_x - cx
dy = pos_y - cy
# Island collider is a box; use a slightly inflated square and push
# away from the nearest face/corner.
inflated = 9.0
if abs(dx) <= inflated and abs(dy) <= inflated:
if abs(dx) > abs(dy):
record(math.copysign(inflated - abs(dx) + 1.0, dx), 0.0, inflated + 1.0, 2.5)
else:
record(0.0, math.copysign(inflated - abs(dy) + 1.0, dy), inflated + 1.0, 2.5)
if best is None:
return None
return best[1], best[2]
def choose_live_input_direction(
seed,
pos_x,
pos_y,
target_x,
target_y,
base_x,
base_y,
live_count,
lookahead,
include_rocks=True,
):
target_ux, target_uy, _ = unit_towards(pos_x, pos_y, target_x, target_y)
if live_count <= 0:
return target_ux, target_uy, False
chunk_x, chunk_y = world_to_chunk(pos_x, pos_y)
radius_chunks = int(math.ceil(lookahead / 16.0)) + 3
obstacles = []
region_x = math.floor(chunk_x / 4)
region_y = math.floor(chunk_y / 4)
region_radius = radius_chunks // 4 + 2
for ry in range(region_y - region_radius, region_y + region_radius + 1):
for rx in range(region_x - region_radius, region_x + region_radius + 1):
sx, sy = structure_chunk(seed, rx, ry)
obstacles.append((sx * 16.0 + 8.0, sy * 16.0 + 8.0, 8.6, 4.0))
if include_rocks:
for cy in range(chunk_y - radius_chunks, chunk_y + radius_chunks + 1):
for cx in range(chunk_x - radius_chunks, chunk_x + radius_chunks + 1):
ox, oy = rock_center(seed, cx, cy)
obstacles.append((ox, oy, 2.8, 1.0))
candidates = [(target_ux, target_uy)]
base_angle = math.atan2(target_uy, target_ux)
for step in range(32):
angle = base_angle + (step / 32.0) * math.tau
candidates.append((math.cos(angle), math.sin(angle)))
best = None
for input_x, input_y in candidates:
wind_x = base_x + input_x * live_count
wind_y = base_y + input_y * live_count
wind_mag = vector_len(wind_x, wind_y)
if wind_mag < 0.1:
continue
path_x = wind_x / wind_mag
path_y = wind_y / wind_mag
progress = wind_x * target_ux + wind_y * target_uy
if progress <= 0.0:
continue
side_x, side_y = -path_y, path_x
risk = 0.0
blocked = False
for ox, oy, clearance, weight in obstacles:
rel_x = ox - pos_x
rel_y = oy - pos_y
ahead = rel_x * path_x + rel_y * path_y
if ahead <= 0.0 or ahead > lookahead:
continue
lateral = abs(rel_x * side_x + rel_y * side_y)
margin = clearance + 1.2
if lateral >= margin:
continue
overlap = (margin - lateral) / margin
near = (lookahead - ahead + 20.0) / (lookahead + 20.0)
risk += overlap * overlap * near * weight
if ahead < 18.0 and lateral < clearance:
blocked = True
target_alignment = path_x * target_ux + path_y * target_uy
score = progress * 1.6 + target_alignment * 8.0 - risk * 130.0
if blocked:
score -= 500.0
if best is None or score > best[0]:
best = (score, input_x, input_y, risk)
if best is None:
return target_ux, target_uy, False
_, input_x, input_y, risk = best
adjusted = risk > 0.05 or input_x * target_ux + input_y * target_uy < 0.95
return input_x, input_y, adjusted
def add_one_wind(url, direction_x, direction_y, pre_send_seconds, settle_seconds):
helper = NgoClient(url, timeout=0.4)
try:
helper.connect_quick_sync()
time.sleep(pre_send_seconds)
helper.send_move(helper.player_object_id, direction_x, direction_y)
time.sleep(settle_seconds)
finally:
helper.close()
def add_wind(
url,
direction_x,
direction_y,
count,
observer=None,
batch_size=3,
pre_send_seconds=0.12,
settle_seconds=0.035,
):
remaining = int(max(0, count))
while remaining:
group = min(max(1, batch_size), remaining)
with concurrent.futures.ThreadPoolExecutor(max_workers=group) as executor:
futures = [
executor.submit(add_one_wind, url, direction_x, direction_y, pre_send_seconds, settle_seconds)
for _ in range(group)
]
for future in concurrent.futures.as_completed(futures):
future.result()
remaining -= group
if observer:
flag = observer.pump_state(0.015)
if flag:
return flag
time.sleep(0.03)
return None
def steer_global_wind(
client,
url,
desired_x,
desired_y,
label,
batch_size=3,
pre_send_seconds=0.12,
settle_seconds=0.035,
):
client.pump_state(0.2)
wind_x, wind_y = client.wind
diff_x = desired_x - wind_x
diff_y = desired_y - wind_y
diff = vector_len(diff_x, diff_y)
if diff < 0.75:
return
count = int(math.ceil(diff))
unit_x = diff_x / diff
unit_y = diff_y / diff
print(
f"[+] wind {label}: current=({wind_x:.1f},{wind_y:.1f}) "
f"target=({desired_x:.1f},{desired_y:.1f}) add={count}",
flush=True,
)
flag = add_wind(
url,
unit_x,
unit_y,
count,
observer=client,
batch_size=batch_size,
pre_send_seconds=pre_send_seconds,
settle_seconds=settle_seconds,
)
if flag:
return flag
client.pump_state(0.5)
# The server stores a sum of unit vectors; keep a local estimate even if the
# latest NetworkVariable delta was coalesced away.
if vector_len(client.wind[0] - wind_x, client.wind[1] - wind_y) < 0.1:
client.wind = (wind_x + unit_x * count, wind_y + unit_y * count)
print(f"[+] wind now approximately ({client.wind[0]:.1f},{client.wind[1]:.1f})", flush=True)
return None
def pulse_global_wind(
client,
url,
direction_x,
direction_y,
count,
label,
batch_size=3,
pre_send_seconds=0.12,
settle_seconds=0.035,
):
unit_x, unit_y = normalize(direction_x, direction_y)
count = int(max(0, round(count)))
if count <= 0 or vector_len(unit_x, unit_y) < 0.5:
return None
client.pump_state(0.05)
old_x, old_y = client.wind
print(
f"[+] pulse {label}: current=({old_x:.1f},{old_y:.1f}) "
f"dir=({unit_x:.2f},{unit_y:.2f}) add={count}",
flush=True,
)
flag = add_wind(
url,
unit_x,
unit_y,
count,
observer=client,
batch_size=batch_size,
pre_send_seconds=pre_send_seconds,
settle_seconds=settle_seconds,
)
if flag:
return flag
client.pump_state(0.2)
if vector_len(client.wind[0] - old_x, client.wind[1] - old_y) < 0.1:
client.wind = (old_x + unit_x * count, old_y + unit_y * count)
print(f"[+] wind now approximately ({client.wind[0]:.1f},{client.wind[1]:.1f})", flush=True)
return None
def seed_scan_waypoints(max_waypoints):
points = []
for y in (8, 72, -56, 136, -120, 200, -184):
for x in (8, 72, 136, -56, -120, 200, -184):
points.append((float(x), float(y)))
if len(points) >= max_waypoints:
return points
return points
def recover_seed_low_from_server(args):
probe = NgoClient(args.url, timeout=0.5)
try:
probe.connect_and_sync()
probe.pump_state(1.0)
candidates = seed_low_candidates_from_chunks(probe.chunks)
print(
f"[+] seed-low scan: chunks={len(probe.chunks)} "
f"structures={sum(1 for chunk in probe.chunks.values() if chunk['has_structure'])} "
f"candidates={len(candidates)}",
flush=True,
)
if len(candidates) == 1:
return candidates[0]
for idx, (target_x, target_y) in enumerate(seed_scan_waypoints(args.seed_scan_max_waypoints), start=1):
pos_x, pos_y = probe.position
ux, uy, _ = unit_towards(pos_x, pos_y, target_x, target_y)
flag = steer_global_wind(
probe,
args.url,
ux * args.seed_scan_wind_mag,
uy * args.seed_scan_wind_mag,
f"seed-scan/{idx}",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
raise RuntimeError("received flag during seed scan before target discovery")
deadline = time.time() + args.seed_scan_waypoint_seconds
while time.time() < deadline:
flag = probe.pump_state(0.08)
if flag:
print(flag, flush=True)
raise RuntimeError("received flag during seed scan before target discovery")
_, _, dist = unit_towards(probe.position[0], probe.position[1], target_x, target_y)
if dist <= args.seed_scan_radius:
break
candidates = seed_low_candidates_from_chunks(probe.chunks)
print(
f"[+] seed-low scan waypoint {idx}: pos=({probe.position[0]:.1f},{probe.position[1]:.1f}) "
f"chunks={len(probe.chunks)} "
f"structures={sum(1 for chunk in probe.chunks.values() if chunk['has_structure'])} "
f"candidates={len(candidates)}",
flush=True,
)
if len(candidates) == 1:
return candidates[0]
raise RuntimeError(f"seed-low scan did not converge: {len(candidates)} candidates remain")
finally:
probe.close()
def discover_target_from_map(args, supplied_seed=None):
print(f"[+] fetching map from {args.map_url}", flush=True)
png_data = fetch_url(args.map_url)
if supplied_seed is None:
seed_for_structures = recover_seed_low_from_server(args)
full_seed = False
print(f"[+] recovered seed low18={seed_for_structures}", flush=True)
else:
seed_for_structures = supplied_seed
full_seed = True
print(f"[+] using supplied seed={seed_for_structures}", flush=True)
flag_chunk_x, flag_chunk_y, region_x, region_y = find_flag_chunk_from_map(seed_for_structures, png_data)
target_x = flag_chunk_x * 16.0 + 8.0
target_y = flag_chunk_y * 16.0 + 8.0
print(
f"[+] map matched flag chunk=({flag_chunk_x},{flag_chunk_y}) "
f"region=({region_x},{region_y}) target=({target_x:.2f},{target_y:.2f})",
flush=True,
)
return target_x, target_y, seed_for_structures, full_seed
def send_live_inputs(clients, x, y):
for client in clients:
client.send_move(client.player_object_id, x, y)
def pump_helpers(clients, duration=0.002):
for client in clients:
client.poll_flag(duration)
def connect_live_helpers(url, count):
helpers = []
for idx in range(count):
helper = NgoClient(url, timeout=0.5)
try:
helper.connect_quick_sync()
except Exception:
helper.close()
raise
helpers.append(helper)
print(
f"[+] live helper {idx + 1}/{count} id={helper.owner_client_id} "
f"player_oid={helper.player_object_id}",
flush=True,
)
return helpers
def run_live_steering(args, target_x, target_y, seed):
clients = []
helpers = []
try:
main_client = NgoClient(args.url)
main_client.connect_and_sync()
clients.append(main_client)
print(f"[+] main client id={main_client.owner_client_id} player_oid={main_client.player_object_id}", flush=True)
if seed is not None:
print(f"[+] seed={seed}", flush=True)
print(f"[+] target=({target_x:.2f}, {target_y:.2f})", flush=True)
ux, uy, _ = unit_towards(main_client.position[0], main_client.position[1], target_x, target_y)
flag = steer_global_wind(
main_client,
args.url,
ux * args.live_base_wind_mag,
uy * args.live_base_wind_mag,
"live-base",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
base_x, base_y = main_client.wind
helpers = connect_live_helpers(args.url, args.live_helpers)
clients.extend(helpers)
live_clients = [main_client] + helpers
live_count = max(1, len(live_clients))
started = time.time()
samples = []
avoid_until = 0.0
avoid_sign = 1.0
helper_pump_at = 0.0
last_status = 0.0
last_heavy_recovery = 0.0
detour_until = 0.0
detour_input = (0.0, 0.0)
detour_sign = 1.0
final_retarget_at = 0.0
final_mode = False
while time.time() - started < args.max_seconds:
now = time.time()
flag = main_client.pump_state(args.live_tick_seconds)
if flag:
print(flag, flush=True)
return 0
pos_x, pos_y = main_client.position
ux, uy, dist = unit_towards(pos_x, pos_y, target_x, target_y)
wind_x, wind_y = main_client.wind
wind_mag = vector_len(wind_x, wind_y)
samples.append((now, pos_x, pos_y, dist))
samples = [sample for sample in samples if now - sample[0] <= max(8.0, args.live_stuck_window + 1.0)]
if now >= helper_pump_at:
pump_helpers(helpers, 0.001)
helper_pump_at = now + 1.0
if now - last_status > args.live_status_interval:
print(
f"[+] t={now-started:.1f}s pos=({pos_x:.1f},{pos_y:.1f}) "
f"dist={dist:.1f} wind=({wind_x:.1f},{wind_y:.1f})",
flush=True,
)
last_status = now
if dist < 12.0:
send_live_inputs(live_clients, 0.0, 0.0)
flag = main_client.pump_state(1.0)
if flag:
print(flag, flush=True)
return 0
flag = steer_global_wind(
main_client,
args.url,
0.0,
0.0,
"capture-stop",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
send_live_inputs(live_clients, ux, uy)
flag = main_client.pump_state(1.0)
if flag:
print(flag, flush=True)
return 0
if not final_mode and dist < args.live_final_radius:
print("[+] entering final approach", flush=True)
send_live_inputs(live_clients, 0.0, 0.0)
time.sleep(0.25)
main_client.pump_state(0.3)
for helper in helpers:
helper.close()
helpers.clear()
clients[:] = [main_client]
live_clients = [main_client]
live_count = 1
flag = steer_global_wind(
main_client,
args.url,
ux * args.near_wind_mag,
uy * args.near_wind_mag,
"final",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
base_x, base_y = main_client.wind
final_retarget_at = time.time() + args.live_final_retarget_interval
final_mode = True
samples.clear()
continue
if final_mode and dist > 12.0 and now >= final_retarget_at:
flag = steer_global_wind(
main_client,
args.url,
ux * args.near_wind_mag,
uy * args.near_wind_mag,
"final-track",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
base_x, base_y = main_client.wind
final_retarget_at = time.time() + args.live_final_retarget_interval
samples.clear()
continue
drive_ux, drive_uy = ux, uy
obstacle_adjusted = False
chosen_live_input = None
if seed is not None and not args.ignore_map_avoidance and dist > 20.0 and not final_mode:
lookahead = max(
args.live_min_lookahead,
min(args.live_max_lookahead, max(1.0, wind_mag) * 7.0 * args.live_lookahead_seconds),
)
chosen_x, chosen_y, obstacle_adjusted = choose_live_input_direction(
seed,
pos_x,
pos_y,
target_x,
target_y,
base_x,
base_y,
live_count,
lookahead,
include_rocks=args.avoid_rocks,
)
chosen_live_input = (chosen_x, chosen_y)
drive_ux, drive_uy = chosen_x, chosen_y
old_enough = [sample for sample in samples if now - sample[0] >= args.live_stuck_window]
stuck = False
if old_enough and dist > 18.0:
old_t, old_x, old_y, old_dist = old_enough[0]
elapsed = max(0.1, now - old_t)
moved = vector_len(pos_x - old_x, pos_y - old_y)
progress = old_dist - dist
if final_mode:
stuck = progress < 1.0 and moved < 1.5 and wind_mag > 0.5
else:
low_progress = max(args.live_stuck_min_progress, wind_mag * 7.0 * elapsed * 0.08)
stuck = progress < low_progress and moved < low_progress * 1.8 and wind_mag > 3.0
if stuck and now >= avoid_until:
escape = (
local_escape_direction(seed, pos_x, pos_y, include_rocks=args.avoid_rocks)
if seed is not None
else None
)
if escape:
drive_ux, drive_uy = escape
else:
drive_ux = -uy * avoid_sign
drive_uy = ux * avoid_sign
avoid_sign *= -1.0
avoid_until = now + args.live_escape_seconds
samples.clear()
print(f"[+] live escape dir=({drive_ux:.2f},{drive_uy:.2f})", flush=True)
if (
args.live_heavy_recovery
and now - last_heavy_recovery >= args.live_recovery_cooldown
):
print("[+] heavy recovery: freeing helper slots and re-vectoring wind", flush=True)
send_live_inputs(live_clients, 0.0, 0.0)
time.sleep(0.25)
main_client.pump_state(0.25)
for helper in helpers:
helper.close()
helpers.clear()
clients[:] = [main_client]
live_clients = [main_client]
live_count = 1
recovery_start_x, recovery_start_y = main_client.position
recovery_dirs = [
(drive_ux, drive_uy),
(-drive_ux, -drive_uy),
(-uy, ux),
(uy, -ux),
]
for attempt, (recover_x, recover_y) in enumerate(recovery_dirs, start=1):
recover_x, recover_y = normalize(recover_x, recover_y)
if vector_len(recover_x, recover_y) < 0.5:
continue
recover_mag = args.live_recovery_wind_mag * (1.0 + 0.35 * (attempt - 1))
flag = steer_global_wind(
main_client,
args.url,
recover_x * recover_mag,
recover_y * recover_mag,
f"recover/{attempt}",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
base_x, base_y = main_client.wind
send_live_inputs(live_clients, recover_x, recover_y)
recovery_deadline = time.time() + args.live_recovery_seconds
while time.time() < recovery_deadline:
flag = main_client.pump_state(0.12)
if flag:
print(flag, flush=True)
return 0
moved = vector_len(
main_client.position[0] - recovery_start_x,
main_client.position[1] - recovery_start_y,
)
if moved >= args.live_recovery_min_move:
drive_ux, drive_uy = recover_x, recover_y
break
print(
f"[+] recovery attempt {attempt} moved {moved:.1f}; trying alternate",
flush=True,
)
send_live_inputs(live_clients, 0.0, 0.0)
time.sleep(0.15)
main_client.pump_state(0.25)
pos_x, pos_y = main_client.position
ux, uy, _ = unit_towards(pos_x, pos_y, target_x, target_y)
resume_mag = args.near_wind_mag if final_mode else args.live_base_wind_mag
flag = steer_global_wind(
main_client,
args.url,
ux * resume_mag,
uy * resume_mag,
"resume",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
base_x, base_y = main_client.wind
if final_mode:
live_clients = [main_client]
live_count = 1
final_retarget_at = time.time() + args.live_final_retarget_interval
else:
helpers = connect_live_helpers(args.url, args.live_helpers)
clients.extend(helpers)
live_clients = [main_client] + helpers
live_count = max(1, len(live_clients))
side_x, side_y = -uy, ux
if abs(drive_ux * side_x + drive_uy * side_y) > 0.15:
detour_sign = math.copysign(1.0, drive_ux * side_x + drive_uy * side_y)
detour_input = (side_x * detour_sign, side_y * detour_sign)
detour_sign *= -1.0
detour_until = time.time() + args.live_detour_seconds
samples.clear()
last_heavy_recovery = time.time()
avoid_until = last_heavy_recovery + 0.5
continue
elif now < avoid_until:
drive_ux = -uy * avoid_sign
drive_uy = ux * avoid_sign
elif final_mode:
drive_ux, drive_uy = ux, uy
if now < detour_until and not final_mode:
input_x, input_y = detour_input
elif chosen_live_input is not None and not final_mode:
input_x, input_y = chosen_live_input
elif args.live_fractional_inputs:
desired_mag = args.near_wind_mag if final_mode else args.live_base_wind_mag + live_count
need_x = drive_ux * desired_mag - wind_x
need_y = drive_uy * desired_mag - wind_y
input_x = need_x / live_count
input_y = need_y / live_count
input_mag = vector_len(input_x, input_y)
if input_mag > 1.0:
input_x /= input_mag
input_y /= input_mag
else:
input_x, input_y = drive_ux, drive_uy
if obstacle_adjusted and chosen_live_input is None and now >= avoid_until and not final_mode:
side_x, side_y = -uy, ux
side_dot = input_x * side_x + input_y * side_y
if abs(side_dot) < 0.2:
input_x, input_y = normalize(input_x + side_x * 0.45, input_y + side_y * 0.45)
send_live_inputs(live_clients, input_x, input_y)
raise RuntimeError("timed out before receiving flag")
finally:
for client in clients:
client.close()
def relay_desired_wind(args, pos_x, pos_y, target_x, target_y):
ux, uy, dist = unit_towards(pos_x, pos_y, target_x, target_y)
mag = args.near_wind_mag if dist < args.slow_radius else args.wind_mag
return ux * mag, uy * mag, dist
def run_relay(args, target_x, target_y, seed):
clients = []
try:
frontier = NgoClient(args.url)
frontier.connect_and_sync()
clients.append(frontier)
print(f"[+] relay boat0 id={frontier.owner_client_id} player_oid={frontier.player_object_id}", flush=True)
if seed is not None:
print(f"[+] seed={seed}", flush=True)
print(f"[+] target=({target_x:.2f}, {target_y:.2f})", flush=True)
desired_x, desired_y, _ = relay_desired_wind(
args,
frontier.position[0],
frontier.position[1],
target_x,
target_y,
)
flag = steer_global_wind(
frontier,
args.url,
desired_x,
desired_y,
"relay-cruise",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
started = time.time()
samples = []
handoffs = 0
capture_failures = 0
connected_at = time.time()
last_status = 0.0
last_revector = 0.0
while time.time() - started < args.max_seconds:
now = time.time()
flag = frontier.pump_state(args.live_tick_seconds)
if flag:
print(flag, flush=True)
return 0
pos_x, pos_y = frontier.position
ux, uy, dist = unit_towards(pos_x, pos_y, target_x, target_y)
wind_x, wind_y = frontier.wind
samples.append((now, pos_x, pos_y, dist))
samples = [sample for sample in samples if now - sample[0] <= max(3.0, args.relay_stuck_window + 0.5)]
if now - last_status >= args.live_status_interval:
print(
f"[+] t={now-started:.1f}s relay#{handoffs} pos=({pos_x:.1f},{pos_y:.1f}) "
f"dist={dist:.1f} wind=({wind_x:.1f},{wind_y:.1f})",
flush=True,
)
last_status = now
force_handoff = False
if dist < args.reach_radius:
flag = steer_global_wind(
frontier,
args.url,
0.0,
0.0,
"relay-stop",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
input_dirs = [
(0.0, 0.0),
(ux, uy),
(-uy, ux),
(uy, -ux),
(-ux, -uy),
]
hold_deadline = time.time() + args.capture_hold_seconds
input_idx = 0
while time.time() < hold_deadline:
move_x, move_y = input_dirs[input_idx % len(input_dirs)]
input_idx += 1
frontier.send_move(frontier.player_object_id, move_x, move_y)
flag = frontier.pump_state(0.08)
if flag:
print(flag, flush=True)
return 0
capture_failures += 1
if capture_failures >= args.capture_tunnel_after:
tunnel_x, tunnel_y, _ = unit_towards(
frontier.position[0],
frontier.position[1],
target_x,
target_y,
)
print(
f"[+] capture tunnel: failures={capture_failures} "
f"dir=({tunnel_x:.2f},{tunnel_y:.2f})",
flush=True,
)
flag = steer_global_wind(
frontier,
args.url,
tunnel_x * args.capture_tunnel_wind_mag,
tunnel_y * args.capture_tunnel_wind_mag,
"capture-tunnel",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
tunnel_deadline = time.time() + args.capture_tunnel_seconds
while time.time() < tunnel_deadline:
frontier.send_move(frontier.player_object_id, tunnel_x, tunnel_y)
flag = frontier.pump_state(0.05)
if flag:
print(flag, flush=True)
return 0
flag = steer_global_wind(
frontier,
args.url,
0.0,
0.0,
"capture-tunnel-stop",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
flag = frontier.pump_state(0.4)
if flag:
print(flag, flush=True)
return 0
capture_failures = 0
force_handoff = True
elif dist > args.reach_radius * 4.0:
capture_failures = 0
desired_x, desired_y, _ = relay_desired_wind(args, pos_x, pos_y, target_x, target_y)
desired_mag = args.near_wind_mag if dist < args.slow_radius else args.wind_mag
if not force_handoff and now - last_revector >= args.relay_revector_interval:
flag = steer_global_wind(
frontier,
args.url,
desired_x,
desired_y,
"relay-final" if dist < args.slow_radius else "relay-cruise",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
last_revector = now
old_enough = [sample for sample in samples if now - sample[0] >= args.relay_stuck_window]
stuck = force_handoff
if old_enough and now - connected_at >= args.relay_settle_seconds and dist > args.reach_radius:
old_t, old_x, old_y, old_dist = old_enough[0]
elapsed = max(0.1, now - old_t)
moved = vector_len(pos_x - old_x, pos_y - old_y)
progress = old_dist - dist
min_progress = args.relay_min_progress
if desired_mag > args.near_wind_mag:
min_progress = max(min_progress, desired_mag * 7.0 * elapsed * 0.02)
stuck = moved < args.relay_min_move and progress < min_progress
if not stuck:
continue
anchor_x, anchor_y = frontier.position
old_client = frontier
handoffs += 1
print(
f"[+] relay handoff #{handoffs}: stuck at ({anchor_x:.1f},{anchor_y:.1f}), "
f"dist={dist:.1f}",
flush=True,
)
new_client = NgoClient(args.url)
try:
old_wind = old_client.wind
new_client.connect_quick_sync()
# If the new boat wedges immediately, it may not dirty its position
# variable. Seed it with the old anchor until movement frames arrive.
new_client.position = (anchor_x, anchor_y)
new_client.wind = old_wind
clients.append(new_client)
learn_deadline = time.time() + args.relay_handoff_seconds
while time.time() < learn_deadline:
flag = new_client.pump_state(0.06)
if flag:
print(flag, flush=True)
return 0
flag = old_client.pump_state(0.001)
if flag:
print(flag, flush=True)
return 0
old_client.close()
clients.remove(old_client)
frontier = new_client
connected_at = time.time()
samples.clear()
last_revector = 0.0
print(
f"[+] relay frontier id={frontier.owner_client_id} "
f"player_oid={frontier.player_object_id} pos=({frontier.position[0]:.1f},{frontier.position[1]:.1f})",
flush=True,
)
except Exception:
new_client.close()
raise
raise RuntimeError("timed out before receiving flag")
finally:
for client in clients:
client.close()
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--url", default="ws://127.0.0.1:7777/")
parser.add_argument("--target-x", type=float)
parser.add_argument("--target-y", type=float)
parser.add_argument("--seed", type=int)
parser.add_argument("--from-docker-logs", action="store_true")
parser.add_argument("--from-map", action="store_true")
parser.add_argument("--discover-only", action="store_true")
parser.add_argument("--map-url", default="http://127.0.0.1:7778/map")
parser.add_argument("--wind-mag", type=float, default=120.0)
parser.add_argument("--near-wind-mag", type=float, default=8.0)
parser.add_argument("--avoid-push", type=float, default=120.0)
parser.add_argument("--slow-radius", type=float, default=650.0)
parser.add_argument("--avoid-seconds", type=float, default=1.7)
parser.add_argument("--stuck-window", type=float, default=1.4)
parser.add_argument("--batch-size", type=int, default=3)
parser.add_argument("--pre-send-seconds", type=float, default=0.12)
parser.add_argument("--settle-seconds", type=float, default=0.035)
parser.add_argument("--tunnel-on-stuck", action="store_true")
parser.add_argument("--straight-tunnel", action="store_true")
parser.add_argument("--ignore-map-avoidance", action="store_true")
parser.add_argument("--avoid-rocks", action=argparse.BooleanOptionalAction, default=True)
parser.add_argument("--stuck-escape-mode", choices=("side", "map"), default="map")
parser.add_argument("--additive-steering", action="store_true")
parser.add_argument("--forward-pulse", type=float, default=45.0)
parser.add_argument("--side-pulse", type=float, default=70.0)
parser.add_argument("--additive-interval", type=float, default=0.8)
parser.add_argument("--tunnel-mult", type=float, default=1.8)
parser.add_argument("--max-wind-mag", type=float, default=0.0)
parser.add_argument("--max-seconds", type=float, default=420.0)
parser.add_argument("--live-steering", action="store_true")
parser.add_argument("--relay", action="store_true")
parser.add_argument("--live-helpers", type=int, default=3)
parser.add_argument("--live-base-wind-mag", type=float, default=7.0)
parser.add_argument("--live-final-radius", type=float, default=120.0)
parser.add_argument("--live-tick-seconds", type=float, default=0.08)
parser.add_argument("--live-lookahead-seconds", type=float, default=0.9)
parser.add_argument("--live-min-lookahead", type=float, default=55.0)
parser.add_argument("--live-max-lookahead", type=float, default=240.0)
parser.add_argument("--live-stuck-window", type=float, default=3.5)
parser.add_argument("--live-stuck-min-progress", type=float, default=8.0)
parser.add_argument("--live-escape-seconds", type=float, default=1.4)
parser.add_argument("--live-status-interval", type=float, default=2.0)
parser.add_argument("--live-fractional-inputs", action="store_true")
parser.add_argument("--live-heavy-recovery", action=argparse.BooleanOptionalAction, default=True)
parser.add_argument("--live-recovery-wind-mag", type=float, default=12.0)
parser.add_argument("--live-recovery-seconds", type=float, default=1.8)
parser.add_argument("--live-recovery-cooldown", type=float, default=8.0)
parser.add_argument("--live-recovery-min-move", type=float, default=8.0)
parser.add_argument("--live-detour-seconds", type=float, default=3.5)
parser.add_argument("--live-final-retarget-interval", type=float, default=2.0)
parser.add_argument("--seed-scan-wind-mag", type=float, default=18.0)
parser.add_argument("--seed-scan-waypoint-seconds", type=float, default=3.0)
parser.add_argument("--seed-scan-max-waypoints", type=int, default=18)
parser.add_argument("--seed-scan-radius", type=float, default=18.0)
parser.add_argument("--relay-stuck-window", type=float, default=1.4)
parser.add_argument("--relay-settle-seconds", type=float, default=1.4)
parser.add_argument("--relay-handoff-seconds", type=float, default=0.15)
parser.add_argument("--relay-revector-interval", type=float, default=3.0)
parser.add_argument("--relay-min-move", type=float, default=3.0)
parser.add_argument("--relay-min-progress", type=float, default=8.0)
parser.add_argument("--reach-radius", type=float, default=8.0)
parser.add_argument("--capture-hold-seconds", type=float, default=1.2)
parser.add_argument("--capture-tunnel-after", type=int, default=3)
parser.add_argument("--capture-tunnel-wind-mag", type=float, default=45.0)
parser.add_argument("--capture-tunnel-seconds", type=float, default=0.35)
args = parser.parse_args()
seed = args.seed
seed_is_full = seed is not None
if args.from_docker_logs:
target_x, target_y = read_target_from_docker_logs()
if seed is None:
seed = read_seed_from_docker_logs()
seed_is_full = True
elif args.from_map:
target_x, target_y, seed, seed_is_full = discover_target_from_map(args, seed)
if not seed_is_full:
# The structure PRNG only needs low18 seed bits, but rock placement uses
# SHA256 over the full seed. Skip rock prediction when only low18 is known.
args.avoid_rocks = False
elif args.target_x is not None and args.target_y is not None:
target_x, target_y = args.target_x, args.target_y
else:
parser.error("provide --target-x/--target-y, --from-map, or --from-docker-logs")
if args.discover_only:
return 0
if args.live_steering:
return run_live_steering(args, target_x, target_y, seed)
if args.relay:
return run_relay(args, target_x, target_y, seed)
clients = []
try:
main_client = NgoClient(args.url)
main_client.connect_and_sync()
clients.append(main_client)
print(f"[+] main client id={main_client.owner_client_id} player_oid={main_client.player_object_id}", flush=True)
started = time.time()
samples = []
avoid_until = 0.0
avoid_sign = 1.0
next_retarget = 0.0
last_status = 0.0
tunnel_mag = args.wind_mag
if seed is not None:
print(f"[+] seed={seed}", flush=True)
print(f"[+] target=({target_x:.2f}, {target_y:.2f})", flush=True)
while time.time() - started < args.max_seconds:
now = time.time()
flag = main_client.pump_state(0.15)
if flag:
print(flag, flush=True)
return 0
pos_x, pos_y = main_client.position
ux, uy, dist = unit_towards(pos_x, pos_y, target_x, target_y)
samples.append((now, pos_x, pos_y, dist))
sample_window = max(3.0, args.stuck_window + 0.5)
if args.additive_steering:
sample_window = max(sample_window, 30.0)
samples = [sample for sample in samples if now - sample[0] <= sample_window]
if now - last_status > 2.0:
print(
f"[+] t={now-started:.1f}s pos=({pos_x:.1f},{pos_y:.1f}) "
f"dist={dist:.1f} wind=({main_client.wind[0]:.1f},{main_client.wind[1]:.1f})",
flush=True,
)
last_status = now
if dist < 12.0:
flag = steer_global_wind(
main_client,
args.url,
0.0,
0.0,
"stop",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
flag = main_client.pump_state(1.0)
if flag:
print(flag, flush=True)
return 0
cruise_mag = args.near_wind_mag if dist < args.slow_radius else max(args.wind_mag, tunnel_mag)
drive_ux, drive_uy = ux, uy
obstacle_adjusted = False
if seed is not None and dist > 20.0 and not args.ignore_map_avoidance:
lookahead = max(80.0, min(420.0, cruise_mag * 2.6))
drive_ux, drive_uy, obstacle_adjusted = local_avoidance_direction(
seed,
pos_x,
pos_y,
target_x,
target_y,
ux,
uy,
lookahead,
include_rocks=args.avoid_rocks,
)
old_enough = [sample for sample in samples if now - sample[0] >= args.stuck_window]
stuck = False
if old_enough and dist > 15.0:
old_t, old_x, old_y, old_dist = old_enough[0]
moved = vector_len(pos_x - old_x, pos_y - old_y)
progress = old_dist - dist
wind_mag = vector_len(*main_client.wind)
if args.additive_steering:
elapsed = max(0.1, now - old_t)
low_progress = min(250.0, max(12.0, wind_mag * elapsed * 0.02))
stuck = progress < low_progress and moved < low_progress * 1.5 and wind_mag > 40.0
else:
stuck = moved < 2.0 and progress < 2.0 and wind_mag > 4.0
if stuck and now >= avoid_until:
current_wind_mag = vector_len(*main_client.wind)
perp_x = -uy * avoid_sign
perp_y = ux * avoid_sign
avoid_sign *= -1.0
avoid_until = now + args.avoid_seconds
next_retarget = avoid_until + 0.2
escape = (
local_escape_direction(seed, pos_x, pos_y, include_rocks=args.avoid_rocks)
if seed is not None
else None
)
if args.additive_steering:
desired_x = drive_ux * 0.15 + perp_x
desired_y = drive_uy * 0.15 + perp_y
flag = pulse_global_wind(
main_client,
args.url,
desired_x,
desired_y,
args.side_pulse,
"side",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
samples.clear()
continue
if args.tunnel_on_stuck:
tunnel_mag = max(tunnel_mag * args.tunnel_mult, current_wind_mag * args.tunnel_mult, args.avoid_push)
if args.max_wind_mag > 0.0:
tunnel_mag = min(tunnel_mag, args.max_wind_mag)
if args.straight_tunnel:
desired_x = ux * tunnel_mag
desired_y = uy * tunnel_mag
elif escape:
escape_x, escape_y = escape
desired_x = escape_x * tunnel_mag
desired_y = escape_y * tunnel_mag
else:
desired_x = (ux * 0.65 + perp_x * 0.35) * tunnel_mag
desired_y = (uy * 0.65 + perp_y * 0.35) * tunnel_mag
label = f"tunnel/{tunnel_mag:.0f}"
elif args.stuck_escape_mode == "map" and escape:
escape_x, escape_y = escape
desired_x = escape_x * max(cruise_mag, args.avoid_push)
desired_y = escape_y * max(cruise_mag, args.avoid_push)
label = "escape"
else:
side_mag = max(cruise_mag, args.avoid_push)
desired_x = drive_ux * cruise_mag * 0.10 + perp_x * side_mag
desired_y = drive_uy * cruise_mag * 0.10 + perp_y * side_mag
label = "avoid"
flag = steer_global_wind(
main_client,
args.url,
desired_x,
desired_y,
label,
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
samples.clear()
continue
if now >= next_retarget and now >= avoid_until:
if args.additive_steering and dist >= args.slow_radius:
wind_x, wind_y = main_client.wind
side_x, side_y = -drive_uy, drive_ux
projection = wind_x * drive_ux + wind_y * drive_uy
lateral = wind_x * side_x + wind_y * side_y
desired_x, desired_y = drive_ux, drive_uy
if not obstacle_adjusted and abs(lateral) > args.wind_mag * 0.7:
desired_x, desired_y = normalize(
drive_ux - side_x * math.copysign(0.8, lateral),
drive_uy - side_y * math.copysign(0.8, lateral),
)
pulse = args.forward_pulse
label = "avoid-forward" if obstacle_adjusted else "forward"
if obstacle_adjusted:
pulse = max(pulse, args.side_pulse)
if args.max_wind_mag > 0.0 and vector_len(wind_x, wind_y) > args.max_wind_mag and projection > args.wind_mag:
pulse = 0
flag = pulse_global_wind(
main_client,
args.url,
desired_x,
desired_y,
pulse,
label,
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
next_retarget = now + args.additive_interval
continue
flag = steer_global_wind(
main_client,
args.url,
drive_ux * cruise_mag,
drive_uy * cruise_mag,
"avoid-map" if obstacle_adjusted else "cruise",
batch_size=args.batch_size,
pre_send_seconds=args.pre_send_seconds,
settle_seconds=args.settle_seconds,
)
if flag:
print(flag, flush=True)
return 0
if seed is not None:
next_retarget = now + (0.9 if obstacle_adjusted else 1.4)
else:
next_retarget = now + (1.5 if dist < args.slow_radius else 5.0)
raise RuntimeError("timed out before receiving flag")
finally:
for client in clients:
client.close()
if __name__ == "__main__":
sys.exit(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment