Add vehicle timetable get/set support
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 22s
Continuous Integration / tests-and-coverage (pull_request) Successful in 24s

Timetables have no GameScript API surface, so this implements real
DoCommands over the game port (ClientCommand/ServerCommand) instead of
the Admin GameScript relay used for list_vehicles(): change_timetable(),
autofill_timetable(), set_timetable_start(), and set_vehicle_on_time()
send commands, while get_vehicle_timetable() reconstructs state purely
by observing ServerCommand broadcasts, since no query command exists.

Includes the custom varuint wire codec these commands require, a full
usage guide (docs/TIMETABLES.md), and a worked demo in main.py.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-16 23:21:21 +02:00
parent 0a8271d57c
commit b33869334a
10 changed files with 998 additions and 10 deletions

View File

@@ -5,6 +5,66 @@ from openttd_protocol.wire.tcp import TCPProtocol
from openttd_protocol.wire.read import read_uint8, read_string, read_uint16, read_uint32
from openttd_protocol.wire.exceptions import SocketClosed
def write_varuint(buffer, value):
"""Encode a non-negative integer using OpenTTD's UTF-8-like varuint scheme."""
if value < 0:
raise ValueError("write_varuint requires a non-negative value")
thresholds = [1 << 7, 1 << 14, 1 << 21, 1 << 28, 1 << 35, 1 << 42, 1 << 49, 1 << 56]
for extra, limit in enumerate(thresholds):
if value < limit:
header_ones = (0xFF << (8 - extra)) & 0xFF
header = header_ones | (value >> (extra * 8))
buffer.append(header)
for i in range(extra - 1, -1, -1):
buffer.append((value >> (i * 8)) & 0xFF)
return
buffer.append(0xFF)
for i in range(7, -1, -1):
buffer.append((value >> (i * 8)) & 0xFF)
def read_varuint(data):
"""Decode a varuint written by write_varuint. Returns (value, rest)."""
header = data[0]
mask = 0x80
extra = 0
while header & mask:
extra += 1
mask >>= 1
value = header & (0x7F >> extra)
rest = data[1:]
for i in range(extra):
value = (value << 8) | rest[i]
return value, rest[extra:]
def write_varuint_signed(buffer, value):
"""Encode a signed integer using zigzag + write_varuint."""
zigzag = (value << 1) ^ (-1 if value < 0 else 0)
write_varuint(buffer, zigzag)
def read_varuint_signed(data):
"""Decode a signed varuint written by write_varuint_signed. Returns (value, rest)."""
zigzag, rest = read_varuint(data)
value = (zigzag >> 1) ^ -(zigzag & 1)
return value, rest
class GameCommand(IntEnum):
ChangeTimetable = 174
SetVehicleOnTime = 176
AutofillTimetable = 177
SetTimetableStart = 180
class ModifyTimetableFlags(IntEnum):
WaitTime = 0
TravelTime = 1
TravelSpeed = 2
SetWaitFixed = 3
SetTravelFixed = 4
SetLeaveType = 5
AssignSchedule = 6
class ModifyTimetableCtrlFlag(IntEnum):
ClearField = 1 << 0
class PacketGameType(IntEnum):
ServerFull = 0
ServerBanned = 1
@@ -229,7 +289,25 @@ class OpenTTDProtocol(TCPProtocol):
@staticmethod
def receive_ServerExternalChat(source, data): return {}
@staticmethod
def receive_ServerCommand(source, data): return {}
def receive_ServerCommand(source, data):
company, data = read_uint8(data)
cmd, data = read_uint16(data)
error_msg, data = read_uint16(data)
tile, data = read_uint32(data)
payload_len, data = read_uint16(data)
payload = data[:payload_len]
data = data[payload_len:]
callback, data = read_uint8(data)
callback_param = 0
if callback != 0:
callback_param, data = read_uint32(data)
frame, data = read_uint32(data)
my_cmd, _ = read_uint8(data)
return {
"company": company, "cmd": cmd, "error_msg": error_msg, "tile": tile,
"payload": payload, "callback": callback, "callback_param": callback_param,
"frame": frame, "my_cmd": bool(my_cmd)
}
@staticmethod
def receive_ServerFull(source, data): return {}
@staticmethod