kovagoadi
2eea541158
Editing (game port, OpenTTDClient): a core of JGRPP's scheduled dispatch DoCommands — set_scheduled_dispatch (enable/disable), add/remove schedule, add/remove/clear slots, and set duration/start date. Adds the command IDs to protocol.py. Viewing (admin, OpenTTDAdminClient.get_dispatch): the GameScript API has no dispatch support, so a new server patch (docker/patches/0002-*) adds read-only GSOrder.GetScheduledDispatch* / IsScheduledDispatchEnabled getters, an AdminBridge GameScript get_dispatch handler exposes them, and get_dispatch() returns the live schedules and slots (mirrors get_timetable). Note: set_dispatch_start_date values are normalised by the engine relative to current game time, so they read back offset from the requested value. Includes unit + e2e tests, a demo in main.py, and protocol/timetable docs. The AdminBridge GameScript and the patched OpenTTD-patches clone live outside this repo; the 0002 patch file is the durable source for the latter. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
840 lines
40 KiB
Python
840 lines
40 KiB
Python
import asyncio
|
|
import logging
|
|
import uuid
|
|
import monocypher
|
|
import os
|
|
import hashlib
|
|
from openttd_protocol.wire.write import write_init, write_string, write_uint8, write_uint16, write_uint32, write_presend, SEND_TCP_MTU
|
|
from openttd_protocol.wire.read import read_uint8, read_uint16
|
|
from .protocol import (
|
|
PacketGameType, OpenTTDProtocol, PacketAdminType, OpenTTDAdminProtocol, NetworkAuthenticationMethod,
|
|
GameCommand, ModifyTimetableFlags, ModifyTimetableCtrlFlag,
|
|
OrderType, OrderStopLocation, INVALID_VEH_ORDER_ID,
|
|
write_varuint, read_varuint, write_varuint_signed, read_varuint_signed
|
|
)
|
|
from .decorators import exclude_call_check
|
|
|
|
class OpenTTDClient:
|
|
"""High-level OpenTTD client for easy integration."""
|
|
def __init__(self, host, port=3979, username="GeminiUser"):
|
|
self.host = host
|
|
self.port = port
|
|
self.username = username
|
|
self.unique_id = str(uuid.uuid4())
|
|
self.log = logging.getLogger(f"OTTDS-{username}")
|
|
|
|
# State
|
|
self.encryption_enabled = False
|
|
self.joined = asyncio.Event()
|
|
self.shutdown_event = asyncio.Event()
|
|
self.client_id = None
|
|
self.vehicle_timetables = {}
|
|
|
|
# Internal crypto
|
|
self._server_password = ""
|
|
self._company_password = ""
|
|
self._target_company = 255
|
|
self._session_key_send = None
|
|
self._session_key_recv = None
|
|
self._encryption_nonce = None
|
|
self._send_aead = None
|
|
self._recv_aead = None
|
|
|
|
# Callbacks
|
|
self.on_chat = None
|
|
|
|
async def connect(self, server_password=""):
|
|
"""Connect to the server and initiate handshake."""
|
|
self._server_password = server_password
|
|
self.log.info(f"Connecting to {self.host}:{self.port}...")
|
|
|
|
loop = asyncio.get_running_loop()
|
|
try:
|
|
self._transport, self._protocol = await loop.create_connection(
|
|
lambda: OpenTTDProtocol(self), self.host, self.port
|
|
)
|
|
d = write_init(PacketGameType.ClientGameInfo)
|
|
write_uint8(d, 4)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
except Exception as e:
|
|
self.log.error(f"Connection failed: {e}")
|
|
raise
|
|
|
|
async def join_company(self, company_id=255, company_password=""):
|
|
"""Join a specific company (0-14, or 255 for spectator)."""
|
|
self._target_company = company_id
|
|
self._company_password = company_password
|
|
if not self.joined.is_set():
|
|
self.log.info(f"Join for company {company_id} configured.")
|
|
else:
|
|
self.log.warning("Already joined.")
|
|
|
|
async def _send_command(self, cmd, payload, tile=0, error_msg=0, callback=0):
|
|
"""Send a DoCommand over the game protocol (ClientCommand packet)."""
|
|
d = write_init(PacketGameType.ClientCommand)
|
|
write_uint8(d, self._target_company)
|
|
write_uint16(d, cmd)
|
|
write_uint16(d, error_msg)
|
|
write_uint32(d, tile)
|
|
write_uint16(d, len(payload))
|
|
d.extend(payload)
|
|
write_uint8(d, callback)
|
|
if callback != 0:
|
|
write_uint32(d, 0)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def change_timetable(self, vehicle_id, order_position, flag, value, clear_field=False):
|
|
"""Change a single order's timetable field (wait/travel time, fixed flags, leave type, ...)."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_uint16(payload, order_position)
|
|
write_uint8(payload, flag)
|
|
write_varuint(payload, value)
|
|
write_uint8(payload, ModifyTimetableCtrlFlag.ClearField if clear_field else 0)
|
|
await self._send_command(GameCommand.ChangeTimetable, payload)
|
|
|
|
async def autofill_timetable(self, vehicle_id, autofill=True, preserve_wait_time=False):
|
|
"""Enable or disable timetable autofill for a vehicle."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_uint8(payload, 1 if autofill else 0)
|
|
write_uint8(payload, 1 if preserve_wait_time else 0)
|
|
await self._send_command(GameCommand.AutofillTimetable, payload)
|
|
|
|
async def set_timetable_start(self, vehicle_id, timetable_all, start_date):
|
|
"""Set the timetable start date for a vehicle (or all vehicles sharing its orders)."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_uint8(payload, 1 if timetable_all else 0)
|
|
write_varuint_signed(payload, start_date)
|
|
await self._send_command(GameCommand.SetTimetableStart, payload)
|
|
|
|
async def set_vehicle_on_time(self, vehicle_id, apply_to_group=False):
|
|
"""Reset a vehicle's lateness counter to make it on-time.
|
|
|
|
This command can only reset lateness to zero; there is no way to mark a vehicle as
|
|
late. If apply_to_group is True, every vehicle sharing this vehicle's order list has
|
|
its lateness reduced by the same amount instead of just this one vehicle. The vehicle's
|
|
timetable must already be running (see set_timetable_start()) or the server rejects
|
|
the command when apply_to_group is False.
|
|
"""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_uint8(payload, 1 if apply_to_group else 0)
|
|
await self._send_command(GameCommand.SetVehicleOnTime, payload)
|
|
|
|
def get_vehicle_timetable(self, vehicle_id):
|
|
"""Return the locally observed timetable state for a vehicle, or None if nothing has been observed.
|
|
|
|
This is a local read with no network round-trip: there is no query command for timetable data in
|
|
the OpenTTD protocol, so this only reflects ServerCommand broadcasts seen since the client joined.
|
|
"""
|
|
return self.vehicle_timetables.get(vehicle_id)
|
|
|
|
async def add_order(self, vehicle_id, station_id, before_position=None, non_stop=0,
|
|
stop_location=OrderStopLocation.PlatformFarEnd, order_flags=0):
|
|
"""Insert a 'go to station' order into a vehicle's order list.
|
|
|
|
By default the new order is appended to the end of the list; pass before_position to insert it
|
|
before an existing order at that index instead. non_stop is an OrderNonStopFlags value
|
|
(0 = stop everywhere) and stop_location an OrderStopLocation value, both packed into the
|
|
order's type byte; stop_location defaults to PlatformFarEnd because the near-end/middle/through
|
|
values are train-only and the server rejects them for other vehicle types. order_flags is the
|
|
16-bit load/unload flag word (0 = the game's defaults: load if possible, unload if possible).
|
|
|
|
Sent over the game port as a real DoCommand: it only succeeds when this client is joined to
|
|
the company that owns the vehicle (see join_company()); a spectator is rejected and kicked.
|
|
"""
|
|
order_type = OrderType.GotoStation | ((stop_location & 0x3) << 4) | ((non_stop & 0x3) << 6)
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_uint16(payload, INVALID_VEH_ORDER_ID if before_position is None else before_position)
|
|
write_uint8(payload, order_type)
|
|
write_uint16(payload, order_flags)
|
|
write_uint16(payload, station_id)
|
|
await self._send_command(GameCommand.InsertOrder, payload)
|
|
|
|
async def remove_order(self, vehicle_id, order_position):
|
|
"""Delete the order at order_position from a vehicle's order list.
|
|
|
|
Sent over the game port as a real DoCommand: like add_order(), it only succeeds when this
|
|
client is joined to the company that owns the vehicle.
|
|
"""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_uint16(payload, order_position)
|
|
await self._send_command(GameCommand.DeleteOrder, payload)
|
|
|
|
# --- Scheduled dispatch (JGRPP) ---
|
|
#
|
|
# A vehicle's order list can hold several dispatch schedules, each with a duration, a start
|
|
# tick and a set of departure slots (offsets within the duration). All of these are edited over
|
|
# the game port and require being joined to the owning company. For an authoritative read of the
|
|
# resulting schedules, use OpenTTDAdminClient.get_dispatch().
|
|
|
|
async def set_scheduled_dispatch(self, vehicle_id, enabled):
|
|
"""Enable or disable scheduled dispatch for a vehicle (and every vehicle sharing its orders)."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_uint8(payload, 1 if enabled else 0)
|
|
await self._send_command(GameCommand.SchDispatch, payload)
|
|
|
|
async def add_dispatch_schedule(self, vehicle_id, start_tick, duration):
|
|
"""Create a new dispatch schedule with the given start tick and duration (in ticks).
|
|
|
|
The schedule is appended to the vehicle's schedule set; its index is the previous schedule
|
|
count (read it back with OpenTTDAdminClient.get_dispatch()). duration must be non-zero.
|
|
"""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_varuint_signed(payload, start_tick)
|
|
write_varuint(payload, duration)
|
|
await self._send_command(GameCommand.SchDispatchAddNewSchedule, payload)
|
|
|
|
async def remove_dispatch_schedule(self, vehicle_id, schedule_index):
|
|
"""Remove the dispatch schedule at schedule_index from a vehicle's schedule set."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_varuint(payload, schedule_index)
|
|
await self._send_command(GameCommand.SchDispatchRemoveSchedule, payload)
|
|
|
|
async def add_dispatch_slot(self, vehicle_id, schedule_index, offset, interval=0, extra_slots=0,
|
|
slot_flags=0, route_id=0):
|
|
"""Add one or more departure slots to a dispatch schedule.
|
|
|
|
offset is the slot's departure time as an offset (in ticks) within the schedule's duration.
|
|
To add several evenly spaced slots in one command, pass extra_slots > 0 together with a
|
|
non-zero interval: each extra slot is placed interval ticks after the previous one (wrapping
|
|
around the duration). slot_flags is the 16-bit slot flag word and route_id an optional
|
|
departure route id (both default to 0).
|
|
"""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_varuint(payload, schedule_index)
|
|
write_varuint(payload, offset)
|
|
write_varuint(payload, interval)
|
|
write_varuint(payload, extra_slots)
|
|
write_uint16(payload, slot_flags)
|
|
write_uint8(payload, route_id)
|
|
await self._send_command(GameCommand.SchDispatchAdd, payload)
|
|
|
|
async def remove_dispatch_slot(self, vehicle_id, schedule_index, offset):
|
|
"""Remove the departure slot at the given offset from a dispatch schedule."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_varuint(payload, schedule_index)
|
|
write_varuint(payload, offset)
|
|
await self._send_command(GameCommand.SchDispatchRemove, payload)
|
|
|
|
async def clear_dispatch_schedule(self, vehicle_id, schedule_index):
|
|
"""Remove every departure slot from a dispatch schedule (leaving the schedule itself)."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_varuint(payload, schedule_index)
|
|
await self._send_command(GameCommand.SchDispatchClear, payload)
|
|
|
|
async def set_dispatch_duration(self, vehicle_id, schedule_index, duration):
|
|
"""Set the total duration (in ticks) of a dispatch schedule."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_varuint(payload, schedule_index)
|
|
write_varuint(payload, duration)
|
|
await self._send_command(GameCommand.SchDispatchSetDuration, payload)
|
|
|
|
async def set_dispatch_start_date(self, vehicle_id, schedule_index, start_tick):
|
|
"""Set the start tick of a dispatch schedule."""
|
|
payload = bytearray()
|
|
write_varuint(payload, vehicle_id)
|
|
write_varuint(payload, schedule_index)
|
|
write_varuint_signed(payload, start_tick)
|
|
await self._send_command(GameCommand.SchDispatchSetStartDate, payload)
|
|
|
|
def disconnect(self, source):
|
|
"""Library callback for when connection is lost."""
|
|
self.log.info("Disconnected.")
|
|
self.shutdown_event.set()
|
|
|
|
async def quit(self):
|
|
"""Gracefully disconnect from the server."""
|
|
if hasattr(self, '_protocol') and not self._transport.is_closing():
|
|
self.log.info("Quitting...")
|
|
try:
|
|
d = write_init(PacketGameType.ClientQuit)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
except Exception:
|
|
pass
|
|
self._transport.close()
|
|
self.shutdown_event.set()
|
|
|
|
# --- Internal Protocol Callbacks ---
|
|
|
|
@exclude_call_check
|
|
def connected(self, source): pass
|
|
|
|
async def receive_ServerGameInfo(self, source, **kwargs):
|
|
self.log.info(f"Server Info: {kwargs.get('name')} ({kwargs.get('openttd_version')})")
|
|
d = write_init(PacketGameType.ClientJoin)
|
|
write_string(d, kwargs.get("openttd_version", "jgrpp-0.71.1"))
|
|
write_uint32(d, 0x20006D64)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerError(self, source, error_code):
|
|
error_names = {
|
|
0: "General",
|
|
1: "Desync",
|
|
2: "SavegameFailed",
|
|
3: "ConnectionLost",
|
|
4: "IllegalPacket",
|
|
5: "NewGRFMismatch",
|
|
6: "NotAuthorized",
|
|
7: "NotExpected",
|
|
8: "WrongRevision",
|
|
9: "NameInUse",
|
|
10: "WrongPassword",
|
|
11: "CompanyMismatch",
|
|
12: "Kicked",
|
|
13: "Cheater",
|
|
14: "ServerFull",
|
|
15: "TooManyCommands",
|
|
16: "TimeoutPassword",
|
|
17: "TimeoutComputer",
|
|
18: "TimeoutMap",
|
|
19: "TimeoutJoin",
|
|
20: "InvalidClientName",
|
|
21: "NotOnAllowList",
|
|
22: "NoAuthenticationMethodAvailable"
|
|
}
|
|
self.log.error(f"Server Error: {error_names.get(error_code, f'Code {error_code}')}")
|
|
await self.quit()
|
|
|
|
async def receive_ServerAuthenticationRequest(self, source, auth_type, data):
|
|
if auth_type == 1:
|
|
server_pub = bytes(data[:32])
|
|
nonce = bytes(data[32:56])
|
|
our_priv = monocypher.generate_key()
|
|
our_pub = monocypher.x25519_public_key(our_priv)
|
|
shared_secret = monocypher.x25519(our_priv, server_pub)
|
|
derived = monocypher.blake2b(shared_secret + server_pub + our_pub + self._server_password.encode())
|
|
self._session_key_send, self._session_key_recv = derived[:32], derived[32:64]
|
|
challenge = os.urandom(8)
|
|
mac, ciphertext = monocypher.lock(self._session_key_send, nonce, challenge, associated_data=our_pub)
|
|
d = write_init(PacketGameType.ClientAuthenticationResponse)
|
|
d.extend(our_pub + mac + ciphertext)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerEnableEncryption(self, source, data):
|
|
self._encryption_nonce = bytes(data)
|
|
self.encryption_enabled = True
|
|
d = write_init(PacketGameType.ClientIdentify)
|
|
write_string(d, self.username)
|
|
write_uint8(d, self._target_company)
|
|
write_uint8(d, 1)
|
|
write_string(d, self.unique_id)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerCheckNewGRFs(self, source):
|
|
d = write_init(PacketGameType.ClientNewGRFsChecked)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerNeedCompanyPassword(self, source, seed, server_id):
|
|
if not self._company_password:
|
|
self.log.error("Server needs company password but none provided.")
|
|
return
|
|
salted = bytearray()
|
|
p_bytes, s_bytes = self._company_password.encode('utf-8'), server_id.encode('utf-8')
|
|
for i in range(32):
|
|
p_char = p_bytes[i] if i < len(p_bytes) else 0
|
|
s_char = s_bytes[i] if i < len(s_bytes) else 0
|
|
seed_char = (seed >> (i % 32)) & 0xFF
|
|
salted.append(p_char ^ s_char ^ seed_char)
|
|
hashed = hashlib.md5(salted, usedforsecurity=False).hexdigest()
|
|
d = write_init(PacketGameType.ClientCompanyPassword)
|
|
write_string(d, hashed)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerWelcome(self, source, **kwargs):
|
|
self.client_id = kwargs.get('client_id')
|
|
self.log.info(f"Successfully joined as client {self.client_id}")
|
|
d = write_init(PacketGameType.ClientGetMap)
|
|
write_uint8(d, 0)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerMapDone(self, source):
|
|
d = write_init(PacketGameType.ClientMapOk)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
self.joined.set()
|
|
|
|
async def receive_ServerFrame(self, source, frame, token):
|
|
d = write_init(PacketGameType.ClientAck)
|
|
write_uint32(d, frame)
|
|
write_uint8(d, token)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerChat(self, source, client_id, message, **kwargs):
|
|
if self.on_chat:
|
|
self.on_chat(client_id, message)
|
|
else:
|
|
self.log.info(f"CHAT: <{client_id}> {message}")
|
|
|
|
async def receive_ServerUnused(self, source, **kwargs): pass
|
|
async def receive_ServerCompanyUpdate(self, source, **kwargs): pass
|
|
async def receive_ServerClientInfo(self, source, **kwargs): pass
|
|
async def receive_ServerSync(self, source, **kwargs): pass
|
|
async def receive_ServerClientJoined(self, source, **kwargs): pass
|
|
async def receive_ServerMapBegin(self, source, **kwargs): pass
|
|
async def receive_ServerMapSize(self, source, **kwargs): pass
|
|
async def receive_ServerMapData(self, source, **kwargs): pass
|
|
async def receive_ServerConfigurationUpdate(self, source, **kwargs): pass
|
|
async def receive_ServerExternalChat(self, source, **kwargs): pass
|
|
_TIMETABLE_FIELD_BY_FLAG = {
|
|
ModifyTimetableFlags.WaitTime: "wait_time",
|
|
ModifyTimetableFlags.TravelTime: "travel_time",
|
|
ModifyTimetableFlags.TravelSpeed: "travel_speed",
|
|
ModifyTimetableFlags.SetWaitFixed: "wait_time_fixed",
|
|
ModifyTimetableFlags.SetTravelFixed: "travel_time_fixed",
|
|
ModifyTimetableFlags.SetLeaveType: "leave_type",
|
|
ModifyTimetableFlags.AssignSchedule: "assigned_schedule",
|
|
}
|
|
_TIMETABLE_BOOL_FLAGS = {ModifyTimetableFlags.SetWaitFixed, ModifyTimetableFlags.SetTravelFixed}
|
|
|
|
async def receive_ServerCommand(self, source, cmd, payload, **kwargs):
|
|
if cmd == GameCommand.ChangeTimetable:
|
|
vehicle_id, rest = read_varuint(payload)
|
|
order_position, rest = read_uint16(rest)
|
|
flag, rest = read_uint8(rest)
|
|
value, rest = read_varuint(rest)
|
|
ctrl_flags, _ = read_uint8(rest)
|
|
entry = self.vehicle_timetables.setdefault(vehicle_id, {"orders": {}})
|
|
order = entry["orders"].setdefault(order_position, {})
|
|
field = self._TIMETABLE_FIELD_BY_FLAG.get(flag)
|
|
if field:
|
|
cleared = bool(ctrl_flags & ModifyTimetableCtrlFlag.ClearField)
|
|
if cleared:
|
|
order[field] = None
|
|
elif flag in self._TIMETABLE_BOOL_FLAGS:
|
|
order[field] = bool(value)
|
|
else:
|
|
order[field] = value
|
|
elif cmd == GameCommand.AutofillTimetable:
|
|
vehicle_id, rest = read_varuint(payload)
|
|
autofill, rest = read_uint8(rest)
|
|
preserve_wait_time, _ = read_uint8(rest)
|
|
entry = self.vehicle_timetables.setdefault(vehicle_id, {"orders": {}})
|
|
entry["autofill"] = bool(autofill)
|
|
entry["autofill_preserve_wait_time"] = bool(preserve_wait_time)
|
|
elif cmd == GameCommand.SetTimetableStart:
|
|
vehicle_id, rest = read_varuint(payload)
|
|
timetable_all, rest = read_uint8(rest)
|
|
start_date, _ = read_varuint_signed(rest)
|
|
entry = self.vehicle_timetables.setdefault(vehicle_id, {"orders": {}})
|
|
entry["timetable_all"] = bool(timetable_all)
|
|
entry["timetable_start"] = start_date
|
|
elif cmd == GameCommand.SetVehicleOnTime:
|
|
vehicle_id, rest = read_varuint(payload)
|
|
apply_to_group, _ = read_uint8(rest)
|
|
entry = self.vehicle_timetables.setdefault(vehicle_id, {"orders": {}})
|
|
entry["on_time_apply_to_group"] = bool(apply_to_group)
|
|
|
|
async def receive_ServerFull(self, source, **kwargs): pass
|
|
async def receive_ServerBanned(self, source, **kwargs): pass
|
|
async def receive_ClientAck(self, source, **kwargs): pass
|
|
async def receive_ClientIdentify(self, source, **kwargs): pass
|
|
|
|
class OpenTTDAdminClient:
|
|
"""High-level OpenTTD Admin client."""
|
|
def __init__(self, host, port=3977, admin_name="GeminiAdmin"):
|
|
self.host = host
|
|
self.port = port
|
|
self.admin_name = admin_name
|
|
self.log = logging.getLogger(f"OTTDA-{admin_name}")
|
|
|
|
# State
|
|
self.encryption_enabled = False
|
|
self.joined = asyncio.Event()
|
|
self.shutdown_event = asyncio.Event()
|
|
|
|
# Internal crypto
|
|
self._admin_password = ""
|
|
self._session_key_send = None
|
|
self._session_key_recv = None
|
|
self._encryption_nonce = None
|
|
self._send_aead = None
|
|
self._recv_aead = None
|
|
|
|
# Callbacks
|
|
self.on_chat = None
|
|
self.on_console = None
|
|
self.on_gamescript = None
|
|
|
|
# GameScript request/response correlation
|
|
self._gs_request_id = 0
|
|
self._gs_futures = {}
|
|
self._gs_subscribed = False
|
|
|
|
async def connect(self, admin_password="", secure=False):
|
|
"""Connect to the admin port and initiate handshake."""
|
|
self._admin_password = admin_password
|
|
self.log.info(f"Connecting to admin {self.host}:{self.port}...")
|
|
|
|
loop = asyncio.get_running_loop()
|
|
try:
|
|
self._transport, self._protocol = await loop.create_connection(
|
|
lambda: OpenTTDAdminProtocol(self), self.host, self.port
|
|
)
|
|
if secure:
|
|
d = write_init(PacketAdminType.AdminJoinSecure)
|
|
write_string(d, self.admin_name)
|
|
write_string(d, "1.0")
|
|
# Bitmask: 1 << NetworkAuthenticationMethod.X25519_PAKE
|
|
write_uint16(d, 1 << NetworkAuthenticationMethod.X25519_PAKE)
|
|
else:
|
|
d = write_init(PacketAdminType.AdminJoin)
|
|
write_string(d, self._admin_password)
|
|
write_string(d, self.admin_name)
|
|
write_string(d, "1.0")
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
except Exception as e:
|
|
self.log.error(f"Admin connection failed: {e}")
|
|
raise
|
|
|
|
def disconnect(self, source):
|
|
"""Library callback for when connection is lost."""
|
|
self.log.info("Admin disconnected.")
|
|
for fut in self._gs_futures.values():
|
|
if not fut.done():
|
|
fut.set_exception(ConnectionError("admin disconnected"))
|
|
self._gs_futures.clear()
|
|
self.shutdown_event.set()
|
|
|
|
async def quit(self):
|
|
"""Gracefully disconnect from the server."""
|
|
if hasattr(self, '_protocol') and not self._transport.is_closing():
|
|
try:
|
|
d = write_init(PacketAdminType.AdminQuit)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
except Exception:
|
|
pass
|
|
self._transport.close()
|
|
self.shutdown_event.set()
|
|
|
|
async def send_rcon(self, command):
|
|
"""Send an RCON command."""
|
|
d = write_init(PacketAdminType.AdminRcon)
|
|
write_string(d, command)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def send_chat(self, message, action=1, dest_type=0, dest_id=0):
|
|
"""Send a chat message as admin. action=1 (CHAT), dest_type=0 (BROADCAST)."""
|
|
d = write_init(PacketAdminType.AdminChat)
|
|
write_uint8(d, action)
|
|
write_uint8(d, dest_type)
|
|
write_uint32(d, dest_id)
|
|
write_string(d, message)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def update_frequency(self, update_type, frequency):
|
|
"""Update the frequency of a certain piece of information."""
|
|
d = write_init(PacketAdminType.AdminUpdateFrequency)
|
|
write_uint16(d, update_type)
|
|
write_uint16(d, frequency)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def poll(self, update_type, data=0xFFFFFFFF):
|
|
"""Poll the server for certain updates."""
|
|
d = write_init(PacketAdminType.AdminPoll)
|
|
write_uint8(d, update_type)
|
|
write_uint32(d, data)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def poll_clients(self, client_id=0xFFFFFFFF):
|
|
"""Poll for client information."""
|
|
from .protocol import AdminUpdateType
|
|
await self.poll(AdminUpdateType.ClientInfo, client_id)
|
|
|
|
async def poll_companies(self, company_id=0xFFFFFFFF):
|
|
"""Poll for company information."""
|
|
from .protocol import AdminUpdateType
|
|
await self.poll(AdminUpdateType.CompanyInfo, company_id)
|
|
|
|
async def poll_economy(self, company_id=0xFFFFFFFF):
|
|
"""Poll for company economy information."""
|
|
from .protocol import AdminUpdateType
|
|
await self.poll(AdminUpdateType.CompanyEconomy, company_id)
|
|
|
|
async def poll_stats(self, company_id=0xFFFFFFFF):
|
|
"""Poll for company statistics."""
|
|
from .protocol import AdminUpdateType
|
|
await self.poll(AdminUpdateType.CompanyStats, company_id)
|
|
|
|
async def list_vehicles(self, company_id=None):
|
|
"""Request a list of vehicles via GameScript. company_id=None for all companies."""
|
|
payload = {"command": "list_vehicles"}
|
|
if company_id is not None:
|
|
payload["company_id"] = company_id
|
|
await self.send_gamescript(payload)
|
|
|
|
async def list_stations(self, company_id=None):
|
|
"""Request a list of stations via GameScript. company_id=None for all companies.
|
|
|
|
Like list_vehicles(), this is fire-and-forget: the AdminBridge GameScript replies with a
|
|
{"stations": [...]} envelope delivered to the on_gamescript callback, so subscribe to
|
|
Gamescript updates first (update_frequency(Gamescript, Automatic)) or the reply is dropped.
|
|
For a station's live cargo detail (waiting vs planned), use get_station().
|
|
"""
|
|
payload = {"command": "list_stations"}
|
|
if company_id is not None:
|
|
payload["company_id"] = company_id
|
|
await self.send_gamescript(payload)
|
|
|
|
async def _gs_query(self, payload, timeout, context):
|
|
"""Send a GameScript request and await its correlated reply.
|
|
|
|
Assigns a fresh request_id, registers a future the ServerGamescript handler resolves when
|
|
the matching reply arrives, and (on first use) subscribes to Gamescript updates so the
|
|
server actually forwards the reply. `payload` is the request dict without request_id;
|
|
`context` is a label used in the ValueError raised on a GameScript-reported error.
|
|
|
|
Raises asyncio.TimeoutError if no reply arrives within `timeout`, ValueError on an error
|
|
reply, and ConnectionError if the admin connection drops while waiting.
|
|
"""
|
|
from .protocol import AdminUpdateType, AdminUpdateFrequency
|
|
if not self._gs_subscribed:
|
|
await self.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
|
|
self._gs_subscribed = True
|
|
self._gs_request_id += 1
|
|
rid = self._gs_request_id
|
|
fut = asyncio.get_running_loop().create_future()
|
|
self._gs_futures[rid] = fut
|
|
request = dict(payload)
|
|
request["request_id"] = rid
|
|
try:
|
|
await self.send_gamescript(request)
|
|
data = await asyncio.wait_for(fut, timeout)
|
|
finally:
|
|
self._gs_futures.pop(rid, None)
|
|
if "error" in data:
|
|
raise ValueError(f"{context}: {data['error']}")
|
|
return data
|
|
|
|
async def get_timetable(self, vehicle_id, timeout=5.0):
|
|
"""Fetch an authoritative timetable snapshot for a vehicle via the AdminBridge GameScript.
|
|
|
|
Unlike the game client's passive observer, this queries the real game state: it works for
|
|
timetables set before this client connected and reflects the actual (not requested) values.
|
|
Auto-subscribes to Gamescript updates on first use; if you manage update frequencies
|
|
yourself, ensure update_frequency(Gamescript, Automatic) is active before calling.
|
|
|
|
Returns a dict with vehicle-level keys (lateness, start_tick, current_order_time,
|
|
total_duration) and an "orders" list of per-order dicts (position, wait_time, travel_time,
|
|
wait_timetabled, travel_timetabled, wait_fixed, travel_fixed, leave_type, max_speed).
|
|
|
|
Raises asyncio.TimeoutError if no reply arrives (e.g. game paused, GS not loaded),
|
|
ValueError on a GameScript-reported error (invalid_vehicle, response_too_large), and
|
|
ConnectionError if the admin connection drops while waiting.
|
|
"""
|
|
return await self._gs_query(
|
|
{"command": "get_timetable", "vehicle_id": vehicle_id}, timeout,
|
|
f"get_timetable({vehicle_id})")
|
|
|
|
async def get_station(self, station_id, timeout=5.0):
|
|
"""Fetch an authoritative snapshot of a station's live cargo state via the AdminBridge GameScript.
|
|
|
|
This queries the real game state (like get_timetable() does for vehicles): it works for any
|
|
existing station regardless of when it was built or when this client connected. Auto-subscribes
|
|
to Gamescript updates on first use; if you manage update frequencies yourself, ensure
|
|
update_frequency(Gamescript, Automatic) is active before calling.
|
|
|
|
Returns a dict with station-level keys (name, location, owner) and a "cargo" list of per-cargo
|
|
dicts. Each cargo dict carries both the real-time and the planned amounts:
|
|
- "waiting": units currently sitting at the station (real-time, GSStation.GetCargoWaiting)
|
|
- "planned": units planned to move through it per the cargodist link graph
|
|
(GSStation.GetCargoPlanned); 0 when cargo distribution is not enabled for that cargo
|
|
- "rating": the station's acceptance rating for the cargo as a percentage (0-100),
|
|
or None if the station has no rating for that cargo yet
|
|
Only cargo types the station has ever handled appear in the list.
|
|
|
|
Raises asyncio.TimeoutError if no reply arrives (e.g. game paused, GS not loaded),
|
|
ValueError on a GameScript-reported error (invalid_station, response_too_large), and
|
|
ConnectionError if the admin connection drops while waiting.
|
|
"""
|
|
return await self._gs_query(
|
|
{"command": "get_station", "station_id": station_id}, timeout,
|
|
f"get_station({station_id})")
|
|
|
|
async def get_station_cargo(self, station_id, cargo_id, from_station=None, via_station=None, timeout=5.0):
|
|
"""Fetch a per-source / per-next-hop breakdown of one cargo at a station via the AdminBridge GS.
|
|
|
|
Where get_station() reports each cargo's totals, this drills into a single cargo type and
|
|
shows how the waiting (real-time) and planned amounts split across the cargo distribution
|
|
(cargodist) link graph. Cargodist tracks every unit by its source station (where it was
|
|
first loaded) and its next hop (the next station it heads to on the way to its final
|
|
destination); there is no separate "final destination" store, so the routing destination is
|
|
the next hop ("via").
|
|
|
|
Returns a dict with the (optionally filtered) totals "waiting" and "planned", plus four
|
|
breakdown lists, each a list of {"station": id, "amount": n} entries (zero amounts omitted):
|
|
- "waiting_by_from" / "planned_by_from": grouped by source station
|
|
- "waiting_by_via" / "planned_by_via": grouped by next hop (routing destination)
|
|
A station id of 65535 (STATION_INVALID) marks cargo whose source was deleted or, as a next
|
|
hop, cargo with no onward routing / to be consumed at this station (also the sole next hop
|
|
for cargo types using manual, non-cargodist distribution).
|
|
|
|
Optional filters narrow the query:
|
|
- from_station: only cargo originating at this source station.
|
|
- via_station: only cargo whose next hop is this station.
|
|
Passing from_station restricts the by_via breakdown to that source (and the totals to it);
|
|
passing via_station restricts the by_from breakdown to that next hop; passing both makes the
|
|
totals the exact source+next-hop amount. Pass 65535 for either to target STATION_INVALID.
|
|
|
|
Raises asyncio.TimeoutError if no reply arrives (e.g. game paused, GS not loaded),
|
|
ValueError on a GameScript-reported error (invalid_station, invalid_cargo,
|
|
response_too_large), and ConnectionError if the admin connection drops while waiting.
|
|
"""
|
|
payload = {"command": "get_station_cargo", "station_id": station_id, "cargo_id": cargo_id}
|
|
if from_station is not None:
|
|
payload["from_station"] = from_station
|
|
if via_station is not None:
|
|
payload["via_station"] = via_station
|
|
return await self._gs_query(payload, timeout, f"get_station_cargo({station_id}, {cargo_id})")
|
|
|
|
async def get_dispatch(self, vehicle_id, timeout=5.0):
|
|
"""Fetch an authoritative snapshot of a vehicle's scheduled dispatch state via the AdminBridge GS.
|
|
|
|
Scheduled dispatch (a JGRPP feature) lets a vehicle depart on a fixed schedule of slots rather
|
|
than purely by timetable. This reads the live state (like get_timetable() does), so it works for
|
|
schedules created before this client connected and reflects the real values. Auto-subscribes to
|
|
Gamescript updates on first use; if you manage update frequencies yourself, ensure
|
|
update_frequency(Gamescript, Automatic) is active before calling.
|
|
|
|
Returns a dict with:
|
|
- "enabled": 1 if scheduled dispatch is turned on for the vehicle, else 0
|
|
- "schedules": a list of per-schedule dicts, each with "index", "duration" (ticks),
|
|
"start_tick", "delay" (max allowed delay), "reuse_slots" (0/1), and "slots" — a list of
|
|
{"offset", "flags"} departure slots (offset is ticks within the schedule duration).
|
|
|
|
These are the same schedules and slots edited by the game-port methods on OpenTTDClient
|
|
(add_dispatch_schedule/add_dispatch_slot/...). Raises asyncio.TimeoutError if no reply arrives
|
|
(e.g. game paused, GS not loaded), ValueError on a GameScript-reported error (invalid_vehicle,
|
|
response_too_large), and ConnectionError if the admin connection drops while waiting.
|
|
"""
|
|
return await self._gs_query(
|
|
{"command": "get_dispatch", "vehicle_id": vehicle_id}, timeout,
|
|
f"get_dispatch({vehicle_id})")
|
|
|
|
async def send_gamescript(self, json_data):
|
|
"""Send a JSON string to the GameScript."""
|
|
import json
|
|
d = write_init(PacketAdminType.AdminGamescript)
|
|
write_string(d, json.dumps(json_data))
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
# --- Internal Protocol Callbacks ---
|
|
|
|
@exclude_call_check
|
|
def connected(self, source): pass
|
|
|
|
async def receive_ServerAuthRequest(self, source, auth_type, data):
|
|
if auth_type == 1: # X25519_PAKE
|
|
server_pub = bytes(data[:32])
|
|
nonce = bytes(data[32:56])
|
|
our_priv = monocypher.generate_key()
|
|
our_pub = monocypher.x25519_public_key(our_priv)
|
|
shared_secret = monocypher.x25519(our_priv, server_pub)
|
|
derived = monocypher.blake2b(shared_secret + server_pub + our_pub + self._admin_password.encode())
|
|
self._session_key_send, self._session_key_recv = derived[:32], derived[32:64]
|
|
challenge = os.urandom(8)
|
|
mac, ciphertext = monocypher.lock(self._session_key_send, nonce, challenge, associated_data=our_pub)
|
|
d = write_init(PacketAdminType.AdminAuthResponse)
|
|
d.extend(our_pub + mac + ciphertext)
|
|
await self._protocol.send_packet(write_presend(d, SEND_TCP_MTU))
|
|
|
|
async def receive_ServerEnableEncryption(self, source, data):
|
|
self._encryption_nonce = bytes(data)
|
|
self.encryption_enabled = True
|
|
self.log.info("Admin encryption enabled.")
|
|
|
|
async def receive_ServerProtocol(self, source, version, updates):
|
|
self.log.info(f"Admin Protocol version {version}")
|
|
|
|
async def receive_ServerWelcome(self, source, **kwargs):
|
|
self.log.info(f"Admin welcomed to {kwargs.get('server_name')}")
|
|
self.joined.set()
|
|
|
|
async def receive_ServerError(self, source, error_code):
|
|
self.log.error(f"Admin Server Error: {error_code}")
|
|
await self.quit()
|
|
|
|
async def receive_ServerChat(self, source, **kwargs):
|
|
if self.on_chat:
|
|
self.on_chat(**kwargs)
|
|
else:
|
|
self.log.info(f"ADMIN CHAT: <{kwargs.get('client_id')}> {kwargs.get('message')}")
|
|
|
|
async def receive_ServerConsole(self, source, **kwargs):
|
|
if self.on_console:
|
|
self.on_console(**kwargs)
|
|
else:
|
|
self.log.info(f"CONSOLE: [{kwargs.get('origin')}] {kwargs.get('text')}")
|
|
|
|
async def receive_ServerRcon(self, source, **kwargs):
|
|
self.log.info(f"RCON: {kwargs.get('text')}")
|
|
|
|
async def receive_ServerRconEnd(self, source, **kwargs):
|
|
self.log.info(f"RCON End: {kwargs.get('command')}")
|
|
|
|
async def receive_ServerClientJoin(self, source, **kwargs):
|
|
self.log.info(f"Admin: Client {kwargs.get('client_id')} joined.")
|
|
|
|
async def receive_ServerClientInfo(self, source, **kwargs):
|
|
self.log.info(f"Admin: Client Info: {kwargs.get('name')} (ID: {kwargs.get('client_id')}, IP: {kwargs.get('network_address')})")
|
|
|
|
async def receive_ServerClientUpdate(self, source, **kwargs):
|
|
self.log.info(f"Admin: Client {kwargs.get('client_id')} updated.")
|
|
|
|
async def receive_ServerClientQuit(self, source, **kwargs):
|
|
self.log.info(f"Admin: Client {kwargs.get('client_id')} quit.")
|
|
|
|
async def receive_ServerClientError(self, source, **kwargs):
|
|
self.log.info(f"Admin: Client {kwargs.get('client_id')} error: {kwargs.get('error_code')}")
|
|
|
|
async def receive_ServerCompanyNew(self, source, **kwargs):
|
|
self.log.info(f"Admin: Company {kwargs.get('company_id')} created.")
|
|
|
|
async def receive_ServerCompanyInfo(self, source, **kwargs):
|
|
self.log.info(f"Admin: Company Info: {kwargs.get('name')} (ID: {kwargs.get('company_id')})")
|
|
|
|
async def receive_ServerCompanyUpdate(self, source, **kwargs):
|
|
self.log.info(f"Admin: Company {kwargs.get('company_id')} updated.")
|
|
|
|
async def receive_ServerCompanyRemove(self, source, **kwargs):
|
|
self.log.info(f"Admin: Company {kwargs.get('company_id')} removed.")
|
|
|
|
async def receive_ServerCompanyEconomy(self, source, **kwargs):
|
|
self.log.info(f"Admin: Company {kwargs.get('company_id')} Economy: Money={kwargs.get('money')}, Loan={kwargs.get('loan')}")
|
|
|
|
async def receive_ServerCompanyStats(self, source, **kwargs):
|
|
self.log.info(f"Admin: Company {kwargs.get('company_id')} Stats: Vehicles={kwargs.get('vehicles')}, Stations={kwargs.get('stations')}")
|
|
|
|
async def receive_ServerGamescript(self, source, **kwargs):
|
|
data = kwargs.get('data')
|
|
if isinstance(data, dict):
|
|
fut = self._gs_futures.get(data.get('request_id'))
|
|
if fut is not None:
|
|
if not fut.done():
|
|
fut.set_result(data)
|
|
return
|
|
if self.on_gamescript:
|
|
self.on_gamescript(data)
|
|
else:
|
|
self.log.info(f"GAMESCRIPT: {data}")
|
|
|
|
async def receive_ServerDate(self, source, **kwargs): pass
|
|
async def receive_ServerFull(self, source, **kwargs): await self.quit()
|
|
async def receive_ServerBanned(self, source, **kwargs): await self.quit()
|
|
async def receive_ServerShutdown(self, source, **kwargs): await self.quit()
|
|
async def receive_ServerNewGame(self, source, **kwargs): pass
|
|
async def receive_ServerPong(self, source, **kwargs):
|
|
self.log.info(f"Admin: Pong received: {kwargs.get('payload')}")
|
|
|
|
|