Compare commits

..

21 Commits

Author SHA1 Message Date
8e352ba248 Merge pull request 'Update debian:13 Docker digest to 34cd9e9' (#23) from renovate/debian-13 into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 23s
Continuous Integration / tests-and-coverage (push) Successful in 28s
Reviewed-on: #23
2026-08-06 17:43:23 +02:00
7fdd5f2fca Merge branch 'main' into renovate/debian-13
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 23s
Continuous Integration / tests-and-coverage (pull_request) Successful in 27s
2026-08-06 17:38:42 +02:00
a9f5b2d5d9 Merge pull request 'Update debian:trixie-slim Docker digest to 3a39a05' (#24) from renovate/debian-trixie-slim into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 49s
Continuous Integration / tests-and-coverage (push) Successful in 30s
Reviewed-on: #24
2026-08-06 17:38:34 +02:00
e925659e28 Update debian:trixie-slim Docker digest to 3a39a05
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 21s
Continuous Integration / tests-and-coverage (pull_request) Successful in 25s
2026-08-06 02:22:05 +00:00
846d085e1a Update debian:13 Docker digest to 34cd9e9
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 1m0s
Continuous Integration / tests-and-coverage (pull_request) Successful in 28s
2026-08-06 02:22:00 +00:00
f0ef4148b0 Merge pull request 'Add scheduled dispatch support (edit + authoritative view)' (#22) from claude/station-realtime-planned-data-0fc51a into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 21s
Continuous Integration / tests-and-coverage (push) Successful in 25s
Reviewed-on: #22
2026-07-24 23:00:14 +02:00
67e886f8d2 Merge branch 'main' into claude/station-realtime-planned-data-0fc51a
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 20s
Continuous Integration / tests-and-coverage (pull_request) Successful in 25s
2026-07-24 22:58:39 +02:00
2eea541158 Add scheduled dispatch support (edit + authoritative view)
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 22s
Continuous Integration / tests-and-coverage (pull_request) Successful in 24s
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>
2026-07-24 22:56:37 +02:00
af1a865fd6 Merge pull request 'Add station listing and cargo queries to admin client' (#21) from claude/station-realtime-planned-data-0fc51a into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 20s
Continuous Integration / tests-and-coverage (push) Successful in 23s
Reviewed-on: #21
2026-07-23 21:52:35 +02:00
81a4d9333d Add station listing and cargo queries to admin client
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 20s
Continuous Integration / tests-and-coverage (pull_request) Successful in 24s
Extends the AdminBridge GameScript JSON channel (the same relay used by
list_vehicles/get_timetable) with station support:

- list_stations(): enumerate stations, fire-and-forget like list_vehicles().
- get_station(): authoritative per-cargo snapshot of a station's live state,
  with both the real-time waiting amount (GSStation.GetCargoWaiting) and the
  planned cargodist link-graph flow (GetCargoPlanned), plus rating.
- get_station_cargo(): break one cargo type down by source station and by
  next hop (the cargodist routing destination) for both waiting and planned
  amounts, with optional from_station/via_station filters.

All three use stock GameScript API (no server patch, unlike timetables).
Refactors the shared GS request/reply correlation out of get_timetable and
get_station into a _gs_query() helper. Companion handlers must be added to
the server-side AdminBridge GameScript (not tracked in this repo).

Includes unit + e2e tests, a worked demo in main_admin.py, and protocol docs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-23 21:48:20 +02:00
c39f970ef9 Merge pull request 'Added real timetable support' (#20) from add-real-timetable-support into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 20s
Continuous Integration / tests-and-coverage (push) Successful in 24s
Reviewed-on: #20
2026-07-23 21:02:37 +02:00
36fc118da3 Merge branch 'main' into add-real-timetable-support
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 26s
Continuous Integration / tests-and-coverage (pull_request) Successful in 24s
2026-07-23 21:00:50 +02:00
3b54a722d6 Added real timetable support
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 33s
Continuous Integration / tests-and-coverage (pull_request) Successful in 26s
2026-07-23 20:59:28 +02:00
38cc5ae40c Merge pull request 'Update actions/setup-python action to v7' (#19) from renovate/actions-setup-python-7.x into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 20s
Continuous Integration / tests-and-coverage (push) Successful in 23s
Reviewed-on: #19
2026-07-21 20:09:28 +02:00
dfb88523f1 Merge branch 'main' into renovate/actions-setup-python-7.x
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 20s
Continuous Integration / tests-and-coverage (pull_request) Successful in 24s
2026-07-21 20:07:54 +02:00
5db0dab9bf Merge pull request 'Update actions/checkout digest to 3d3c42e' (#18) from renovate/actions-checkout-digest into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 22s
Continuous Integration / tests-and-coverage (push) Successful in 26s
Reviewed-on: #18
2026-07-21 20:07:47 +02:00
71bef5a8a2 Update actions/setup-python action to v7
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 1m55s
Continuous Integration / tests-and-coverage (pull_request) Successful in 24s
2026-07-21 02:22:10 +00:00
df9c7b3f06 Update actions/checkout digest to 3d3c42e
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 55s
Continuous Integration / tests-and-coverage (pull_request) Successful in 24s
2026-07-21 02:22:05 +00:00
954663e80c Merge pull request 'Add vehicle timetable get/set support' (#17) from claude/listing-vehicles-support-d07bf0 into main
All checks were successful
Continuous Integration / lint-and-security (push) Successful in 21s
Continuous Integration / tests-and-coverage (push) Successful in 25s
Reviewed-on: #17
2026-07-16 23:26:14 +02:00
6a28e4acff Merge branch 'main' into claude/listing-vehicles-support-d07bf0
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 21s
Continuous Integration / tests-and-coverage (pull_request) Successful in 25s
2026-07-16 23:25:21 +02:00
b33869334a 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>
2026-07-16 23:21:21 +02:00
18 changed files with 2768 additions and 23 deletions

View File

@@ -10,9 +10,9 @@ jobs:
lint-and-security:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
- name: Set up Python
uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
with:
python-version: '3.12'
@@ -32,10 +32,10 @@ jobs:
tests-and-coverage:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
- name: Set up Python
uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
with:
python-version: '3.12'

View File

@@ -10,6 +10,13 @@ A high-performance, Object-Oriented Python client for OpenTTD servers, specifica
- **State Management:** Handles the full join sequence including Map download and synchronization.
- **Comprehensive Testing:** Robustly tested with unit, logic, and E2E tests (including 100% coverage for unit/logic tests).
- **Vehicle Listing:** Query vehicle data via the Admin GameScript channel with `list_vehicles()`.
- **Vehicle Timetables:** Read and modify a vehicle's timetable (`change_timetable()`, `autofill_timetable()`, `set_timetable_start()`, `set_vehicle_on_time()`, `get_vehicle_timetable()`) via real game-protocol commands.
- **Order Editing:** Add and remove a vehicle's orders (`add_order()` inserts a "go to station" stop, `remove_order()` deletes one) via real game-protocol commands.
- **Scheduled Dispatch (JGRPP):** Edit a vehicle's dispatch schedules and departure slots over the game port (`add_dispatch_schedule()`, `add_dispatch_slot()`, `set_dispatch_duration()`, `set_scheduled_dispatch()`, and more), and read them back authoritatively with `OpenTTDAdminClient.get_dispatch()` (via a patched GameScript API + the AdminBridge GS).
- **Authoritative Timetable Reads:** `OpenTTDAdminClient.get_timetable()` fetches the real, current timetable of any vehicle from the running game (via a patched GameScript API + the AdminBridge GS) — no company join needed, works for timetables set before connecting.
- **Station Listing:** Enumerate stations via the Admin GameScript channel with `list_stations()`.
- **Station Cargo Snapshots:** `OpenTTDAdminClient.get_station()` returns a station's live per-cargo state from the running game — both the **real-time** amount waiting and the **planned** flow through the cargodist link graph — over the AdminBridge GS (stock GameScript API, no server patch needed).
- **Cargo Flow Breakdown:** `OpenTTDAdminClient.get_station_cargo()` breaks one cargo type down by **source station** and **next hop** (routing destination) for both waiting (real-time) and planned amounts, with optional `from_station`/`via_station` filters.
## 🛠 Setup
@@ -73,4 +80,5 @@ For detailed instructions on E2E testing and coverage reports, see the [Testing
## 📜 Documentation
- [Architecture & Design](docs/ARCHITECTURE.md)
- [Protocol Internals (PAKE/Encryption)](docs/PROTOCOL.md)
- [Vehicle Timetables Usage Guide](docs/TIMETABLES.md)
- [Contributor Guide](docs/CONTRIBUTING.md)

View File

@@ -1,5 +1,5 @@
# Build stage
FROM debian:13@sha256:fac46bff2e02f51425b6e33b0e1169f55dfb053d83511ca28aa50c09fd5ed7a4 AS builder
FROM debian:13@sha256:34cd9e9fd437c0a095ec39cb2e73422c9f30821b0d0848ed74fd0d43bae4d958 AS builder
RUN apt-get update && apt-get install -y \
build-essential \
@@ -26,13 +26,16 @@ RUN cmake .. \
&& make -j$(nproc) install
# Runtime stage
FROM debian:bookworm-slim@sha256:7b140f374b289a7c2befc338f42ebe6441b7ea838a042bbd5acbfca6ec875818
# Must track the builder's Debian release: the builder (debian:13/trixie) links
# against glibc 2.38+, so an older runtime (e.g. bookworm, glibc 2.36) cannot run
# the resulting binary. Package names use the trixie t64 spelling.
FROM debian:trixie-slim@sha256:3a39a0592364683e6bab97937b72cad5a8fa6dcbbee90edb3bb48c7f8e94f258
RUN apt-get update && apt-get install -y \
libcurl3-gnutls \
libcurl3t64-gnutls \
liblzma5 \
liblzo2-2 \
libpng16-16 \
libpng16-16t64 \
libzstd1 \
zlib1g \
ca-certificates \

View File

@@ -25,13 +25,18 @@ This setup builds OpenTTD with the JGR Patch Pack (JGRPP) from source and runs i
Save games are stored in `config/save/`.
## JGRPP Source
The source code is cloned from the `jgrpp` branch of `https://github.com/JGRennison/OpenTTD-patches`.
The source code is cloned from the `jgrpp` branch of `https://github.com/JGRennison/OpenTTD-patches`
(currently at tag `jgrpp-0.71.1`) and carries local patches from `patches/` — see
[patches/README.md](patches/README.md). After a fresh clone of the source, apply them with
`git -C OpenTTD-patches am ../patches/*.patch` before building.
To update the server to a newer JGRPP version:
1. Update the `OpenTTD-patches` directory:
```bash
cd OpenTTD-patches && git pull && cd ..
```
2. Rebuild the image:
2. Reapply (rebase if needed) the local patches from `patches/`.
3. Rebuild the image:
```bash
docker-compose up -d --build
```

View File

@@ -0,0 +1,257 @@
From 73f6770eb3e0ec5be8da9e74fa0874fb6ba2d36d Mon Sep 17 00:00:00 2001
From: kovagoadi <kovagoadi@gmail.com>
Date: Sun, 19 Jul 2026 00:33:23 +0200
Subject: [PATCH] Add GameScript API timetable getters to ScriptOrder
Expose read-only timetable data to AI/GS scripts: per-order wait/travel
times, timetabled/fixed flags, leave type and max speed, plus per-vehicle
lateness, timetable start tick, current order time and total duration.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
src/script/api/script_order.cpp | 104 ++++++++++++++++++++++++++++
src/script/api/script_order.hpp | 116 ++++++++++++++++++++++++++++++++
2 files changed, 220 insertions(+)
diff --git a/src/script/api/script_order.cpp b/src/script/api/script_order.cpp
index 865df623f9..ee18f7d588 100644
--- a/src/script/api/script_order.cpp
+++ b/src/script/api/script_order.cpp
@@ -718,3 +718,107 @@ static void _DoCommandReturnSetOrderFlags(class ScriptInstance &instance)
return ScriptMap::DistanceManhattan(origin_tile, dest_tile);
}
}
+
+/* static */ SQInteger ScriptOrder::GetTimetableWaitTime(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return -1;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return -1;
+ return order->GetWaitTime();
+}
+
+/* static */ SQInteger ScriptOrder::GetTimetableTravelTime(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return -1;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return -1;
+ return order->GetTravelTime();
+}
+
+/* static */ bool ScriptOrder::IsWaitTimetabled(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return false;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return false;
+ return order->IsWaitTimetabled();
+}
+
+/* static */ bool ScriptOrder::IsTravelTimetabled(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return false;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return false;
+ return order->IsTravelTimetabled();
+}
+
+/* static */ bool ScriptOrder::IsWaitFixed(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return false;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return false;
+ return order->IsWaitFixed();
+}
+
+/* static */ bool ScriptOrder::IsTravelFixed(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return false;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return false;
+ return order->IsTravelFixed();
+}
+
+/* static */ SQInteger ScriptOrder::GetLeaveType(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return -1;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return -1;
+ return order->GetLeaveType();
+}
+
+/* static */ SQInteger ScriptOrder::GetTimetableMaxSpeed(VehicleID vehicle_id, OrderPosition order_position)
+{
+ if (!IsValidVehicleOrder(vehicle_id, order_position)) return -1;
+
+ const Order *order = ::ResolveOrder(vehicle_id, order_position);
+ if (order == nullptr) return -1;
+ return order->GetMaxSpeed();
+}
+
+/* static */ SQInteger ScriptOrder::GetTimetableLateness(VehicleID vehicle_id)
+{
+ if (!ScriptVehicle::IsPrimaryVehicle(vehicle_id)) return 0;
+
+ return ::Vehicle::Get(vehicle_id)->lateness_counter;
+}
+
+/* static */ SQInteger ScriptOrder::GetTimetableStartTick(VehicleID vehicle_id)
+{
+ if (!ScriptVehicle::IsPrimaryVehicle(vehicle_id)) return -1;
+
+ return ::Vehicle::Get(vehicle_id)->timetable_start.base();
+}
+
+/* static */ SQInteger ScriptOrder::GetCurrentOrderTime(VehicleID vehicle_id)
+{
+ if (!ScriptVehicle::IsPrimaryVehicle(vehicle_id)) return -1;
+
+ return ::Vehicle::Get(vehicle_id)->current_order_time;
+}
+
+/* static */ SQInteger ScriptOrder::GetTimetableTotalDuration(VehicleID vehicle_id)
+{
+ if (!ScriptVehicle::IsPrimaryVehicle(vehicle_id)) return -1;
+
+ const Vehicle *v = ::Vehicle::Get(vehicle_id);
+ if (v->orders == nullptr) return -1;
+ Ticks duration = v->orders->GetTimetableTotalDuration();
+ if (duration == INVALID_TICKS) return -1;
+ return duration;
+}
diff --git a/src/script/api/script_order.hpp b/src/script/api/script_order.hpp
index 81dc06cd7d..6c96b91b3a 100644
--- a/src/script/api/script_order.hpp
+++ b/src/script/api/script_order.hpp
@@ -604,6 +604,122 @@ public:
* @see ScriptEngine::GetMaximumOrderDistance and ScriptVehicle::GetMaximumOrderDistance
*/
static SQInteger GetOrderDistance(ScriptVehicle::VehicleType vehicle_type, TileIndex origin_tile, TileIndex dest_tile);
+
+ /**
+ * Gets the timetabled wait time of the given order for the given vehicle.
+ * @param vehicle_id The vehicle to get the timetable wait time for.
+ * @param order_position The order to get the timetable wait time for.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return The wait time of the order in ticks, or -1 when the order is invalid.
+ * @note The raw stored wait time is returned even if the wait time is not
+ * timetabled; use IsWaitTimetabled to check whether it is explicitly set.
+ */
+ static SQInteger GetTimetableWaitTime(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Gets the timetabled travel time of the given order for the given vehicle.
+ * @param vehicle_id The vehicle to get the timetable travel time for.
+ * @param order_position The order to get the timetable travel time for.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return The travel time of the order in ticks, or -1 when the order is invalid.
+ * @note The raw stored travel time is returned even if the travel time is not
+ * timetabled; use IsTravelTimetabled to check whether it is explicitly set.
+ */
+ static SQInteger GetTimetableTravelTime(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Checks whether the wait time of the given order is timetabled (explicitly set).
+ * @param vehicle_id The vehicle to check the order for.
+ * @param order_position The order to check.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return True if and only if the wait time is timetabled.
+ */
+ static bool IsWaitTimetabled(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Checks whether the travel time of the given order is timetabled (explicitly set).
+ * @param vehicle_id The vehicle to check the order for.
+ * @param order_position The order to check.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return True if and only if the travel time is timetabled.
+ */
+ static bool IsTravelTimetabled(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Checks whether the wait time of the given order is fixed (locked against autofill).
+ * @param vehicle_id The vehicle to check the order for.
+ * @param order_position The order to check.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return True if and only if the wait time is fixed.
+ */
+ static bool IsWaitFixed(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Checks whether the travel time of the given order is fixed (locked against autofill).
+ * @param vehicle_id The vehicle to check the order for.
+ * @param order_position The order to check.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return True if and only if the travel time is fixed.
+ */
+ static bool IsTravelFixed(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Gets the leave type of the given order for the given vehicle.
+ * @param vehicle_id The vehicle to get the leave type for.
+ * @param order_position The order to get the leave type for.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return The leave type of the order (0 = leave when timetabled, 1 = leave as
+ * soon as possible, 2 = leave early if any cargo fully loaded, 3 = leave early
+ * if all cargo fully loaded), or -1 when the order is invalid.
+ */
+ static SQInteger GetLeaveType(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Gets the timetabled maximum speed of the given order for the given vehicle.
+ * @param vehicle_id The vehicle to get the timetable max speed for.
+ * @param order_position The order to get the timetable max speed for.
+ * @pre IsValidVehicleOrder(vehicle_id, order_position).
+ * @return The maximum speed of the order (65535 when no speed cap is set),
+ * or -1 when the order is invalid.
+ */
+ static SQInteger GetTimetableMaxSpeed(VehicleID vehicle_id, OrderPosition order_position);
+
+ /**
+ * Gets the timetable lateness of the given vehicle.
+ * @param vehicle_id The vehicle to get the lateness for.
+ * @pre ScriptVehicle::IsPrimaryVehicle(vehicle_id).
+ * @return How many ticks the vehicle is late; negative values mean the vehicle
+ * is running early. Returns 0 when the vehicle is invalid, which is
+ * indistinguishable from an on-time vehicle; check the vehicle validity first.
+ */
+ static SQInteger GetTimetableLateness(VehicleID vehicle_id);
+
+ /**
+ * Gets the state tick at which the timetable of the given vehicle starts.
+ * @param vehicle_id The vehicle to get the timetable start tick for.
+ * @pre ScriptVehicle::IsPrimaryVehicle(vehicle_id).
+ * @return The absolute state tick the timetable starts at (0 when the
+ * timetable has not been started), or -1 when the vehicle is invalid.
+ */
+ static SQInteger GetTimetableStartTick(VehicleID vehicle_id);
+
+ /**
+ * Gets the number of ticks the given vehicle has spent on its current order.
+ * @param vehicle_id The vehicle to get the current order time for.
+ * @pre ScriptVehicle::IsPrimaryVehicle(vehicle_id).
+ * @return The number of ticks spent on the current order, or -1 when the
+ * vehicle is invalid.
+ */
+ static SQInteger GetCurrentOrderTime(VehicleID vehicle_id);
+
+ /**
+ * Gets the total duration of the timetable of the given vehicle.
+ * @param vehicle_id The vehicle to get the timetable duration for.
+ * @pre ScriptVehicle::IsPrimaryVehicle(vehicle_id).
+ * @return The total timetable duration in ticks, or -1 when the vehicle is
+ * invalid, has no orders, or the timetable is not complete.
+ */
+ static SQInteger GetTimetableTotalDuration(VehicleID vehicle_id);
};
DECLARE_ENUM_AS_BIT_SET(ScriptOrder::ScriptOrderFlags)
--
2.54.0

View File

@@ -0,0 +1,195 @@
From 1e4bdcca84e956ece32d2d77dc8001bfd1d8e2f8 Mon Sep 17 00:00:00 2001
From: kovagoadi <kovagoadi@gmail.com>
Date: Fri, 24 Jul 2026 22:32:29 +0200
Subject: [PATCH] Add GameScript API scheduled dispatch getters to ScriptOrder
Expose read-only scheduled dispatch data to AI/GS scripts: per-vehicle
schedule count and enabled flag, per-schedule duration, start tick, max
delay and slot re-use, and per-slot offset and flags. Enables the
AdminBridge GameScript's get_dispatch command and the Python client's
OpenTTDAdminClient.get_dispatch().
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
src/script/api/script_order.cpp | 81 +++++++++++++++++++++++++++++++++
src/script/api/script_order.hpp | 75 ++++++++++++++++++++++++++++++
2 files changed, 156 insertions(+)
diff --git a/src/script/api/script_order.cpp b/src/script/api/script_order.cpp
index ee18f7d588..3db4639f14 100644
--- a/src/script/api/script_order.cpp
+++ b/src/script/api/script_order.cpp
@@ -822,3 +822,84 @@ static void _DoCommandReturnSetOrderFlags(class ScriptInstance &instance)
if (duration == INVALID_TICKS) return -1;
return duration;
}
+
+/**
+ * Resolve a scheduled dispatch schedule for a vehicle, or nullptr if the vehicle/schedule is invalid.
+ */
+static const DispatchSchedule *ResolveDispatchSchedule(VehicleID vehicle_id, SQInteger schedule_index)
+{
+ if (!ScriptVehicle::IsPrimaryVehicle(vehicle_id)) return nullptr;
+ const Vehicle *v = ::Vehicle::Get(vehicle_id);
+ if (v->orders == nullptr) return nullptr;
+ if (schedule_index < 0 || static_cast<uint>(schedule_index) >= v->orders->GetScheduledDispatchScheduleCount()) return nullptr;
+ return &v->orders->GetDispatchScheduleByIndex(static_cast<uint>(schedule_index));
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchScheduleCount(VehicleID vehicle_id)
+{
+ if (!ScriptVehicle::IsPrimaryVehicle(vehicle_id)) return -1;
+
+ const Vehicle *v = ::Vehicle::Get(vehicle_id);
+ if (v->orders == nullptr) return 0;
+ return v->orders->GetScheduledDispatchScheduleCount();
+}
+
+/* static */ SQInteger ScriptOrder::IsScheduledDispatchEnabled(VehicleID vehicle_id)
+{
+ if (!ScriptVehicle::IsPrimaryVehicle(vehicle_id)) return -1;
+
+ return ::Vehicle::Get(vehicle_id)->vehicle_flags.Test(VehicleFlag::ScheduledDispatch) ? 1 : 0;
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchDuration(VehicleID vehicle_id, SQInteger schedule_index)
+{
+ const DispatchSchedule *ds = ::ResolveDispatchSchedule(vehicle_id, schedule_index);
+ if (ds == nullptr) return -1;
+ return ds->GetScheduledDispatchDuration();
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchStartTick(VehicleID vehicle_id, SQInteger schedule_index)
+{
+ const DispatchSchedule *ds = ::ResolveDispatchSchedule(vehicle_id, schedule_index);
+ if (ds == nullptr) return -1;
+ return ds->GetScheduledDispatchStartTick().base();
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchDelay(VehicleID vehicle_id, SQInteger schedule_index)
+{
+ const DispatchSchedule *ds = ::ResolveDispatchSchedule(vehicle_id, schedule_index);
+ if (ds == nullptr) return -1;
+ return ds->GetScheduledDispatchDelay();
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchReuseSlots(VehicleID vehicle_id, SQInteger schedule_index)
+{
+ const DispatchSchedule *ds = ::ResolveDispatchSchedule(vehicle_id, schedule_index);
+ if (ds == nullptr) return -1;
+ return ds->GetScheduledDispatchReuseSlots() ? 1 : 0;
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchSlotCount(VehicleID vehicle_id, SQInteger schedule_index)
+{
+ const DispatchSchedule *ds = ::ResolveDispatchSchedule(vehicle_id, schedule_index);
+ if (ds == nullptr) return -1;
+ return (SQInteger)ds->GetScheduledDispatch().size();
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchSlotOffset(VehicleID vehicle_id, SQInteger schedule_index, SQInteger slot_index)
+{
+ const DispatchSchedule *ds = ::ResolveDispatchSchedule(vehicle_id, schedule_index);
+ if (ds == nullptr) return -1;
+ const std::vector<DispatchSlot> &slots = ds->GetScheduledDispatch();
+ if (slot_index < 0 || static_cast<size_t>(slot_index) >= slots.size()) return -1;
+ return slots[static_cast<size_t>(slot_index)].offset;
+}
+
+/* static */ SQInteger ScriptOrder::GetScheduledDispatchSlotFlags(VehicleID vehicle_id, SQInteger schedule_index, SQInteger slot_index)
+{
+ const DispatchSchedule *ds = ::ResolveDispatchSchedule(vehicle_id, schedule_index);
+ if (ds == nullptr) return -1;
+ const std::vector<DispatchSlot> &slots = ds->GetScheduledDispatch();
+ if (slot_index < 0 || static_cast<size_t>(slot_index) >= slots.size()) return -1;
+ return slots[static_cast<size_t>(slot_index)].flags;
+}
diff --git a/src/script/api/script_order.hpp b/src/script/api/script_order.hpp
index 6c96b91b3a..d0a7e41fa7 100644
--- a/src/script/api/script_order.hpp
+++ b/src/script/api/script_order.hpp
@@ -720,6 +720,81 @@ public:
* invalid, has no orders, or the timetable is not complete.
*/
static SQInteger GetTimetableTotalDuration(VehicleID vehicle_id);
+
+ /**
+ * Gets the number of scheduled dispatch schedules of the given vehicle.
+ * @param vehicle_id The vehicle to query.
+ * @pre ScriptVehicle::IsPrimaryVehicle(vehicle_id).
+ * @return The number of dispatch schedules (0 when the vehicle has no order list),
+ * or -1 when the vehicle is invalid.
+ */
+ static SQInteger GetScheduledDispatchScheduleCount(VehicleID vehicle_id);
+
+ /**
+ * Gets whether scheduled dispatch is enabled for the given vehicle.
+ * @param vehicle_id The vehicle to query.
+ * @pre ScriptVehicle::IsPrimaryVehicle(vehicle_id).
+ * @return 1 if enabled, 0 if disabled, or -1 when the vehicle is invalid.
+ */
+ static SQInteger IsScheduledDispatchEnabled(VehicleID vehicle_id);
+
+ /**
+ * Gets the duration in ticks of a dispatch schedule.
+ * @param vehicle_id The vehicle to query.
+ * @param schedule_index The dispatch schedule index.
+ * @return The schedule duration in ticks, or -1 when the vehicle or schedule is invalid.
+ */
+ static SQInteger GetScheduledDispatchDuration(VehicleID vehicle_id, SQInteger schedule_index);
+
+ /**
+ * Gets the start tick of a dispatch schedule.
+ * @param vehicle_id The vehicle to query.
+ * @param schedule_index The dispatch schedule index.
+ * @return The absolute start state tick, or -1 when the vehicle or schedule is invalid.
+ */
+ static SQInteger GetScheduledDispatchStartTick(VehicleID vehicle_id, SQInteger schedule_index);
+
+ /**
+ * Gets the maximum allowed delay of a dispatch schedule.
+ * @param vehicle_id The vehicle to query.
+ * @param schedule_index The dispatch schedule index.
+ * @return The maximum delay in ticks, or -1 when the vehicle or schedule is invalid.
+ */
+ static SQInteger GetScheduledDispatchDelay(VehicleID vehicle_id, SQInteger schedule_index);
+
+ /**
+ * Gets whether a dispatch schedule re-uses its dispatch slots.
+ * @param vehicle_id The vehicle to query.
+ * @param schedule_index The dispatch schedule index.
+ * @return 1 if slots are re-used, 0 if not, or -1 when the vehicle or schedule is invalid.
+ */
+ static SQInteger GetScheduledDispatchReuseSlots(VehicleID vehicle_id, SQInteger schedule_index);
+
+ /**
+ * Gets the number of departure slots in a dispatch schedule.
+ * @param vehicle_id The vehicle to query.
+ * @param schedule_index The dispatch schedule index.
+ * @return The number of slots, or -1 when the vehicle or schedule is invalid.
+ */
+ static SQInteger GetScheduledDispatchSlotCount(VehicleID vehicle_id, SQInteger schedule_index);
+
+ /**
+ * Gets the departure offset (in ticks, within the schedule duration) of a dispatch slot.
+ * @param vehicle_id The vehicle to query.
+ * @param schedule_index The dispatch schedule index.
+ * @param slot_index The slot index within the schedule.
+ * @return The slot offset, or -1 when the vehicle, schedule or slot is invalid.
+ */
+ static SQInteger GetScheduledDispatchSlotOffset(VehicleID vehicle_id, SQInteger schedule_index, SQInteger slot_index);
+
+ /**
+ * Gets the flag word of a dispatch slot.
+ * @param vehicle_id The vehicle to query.
+ * @param schedule_index The dispatch schedule index.
+ * @param slot_index The slot index within the schedule.
+ * @return The slot flags, or -1 when the vehicle, schedule or slot is invalid.
+ */
+ static SQInteger GetScheduledDispatchSlotFlags(VehicleID vehicle_id, SQInteger schedule_index, SQInteger slot_index);
};
DECLARE_ENUM_AS_BIT_SET(ScriptOrder::ScriptOrderFlags)
--
2.54.0

36
docker/patches/README.md Normal file
View File

@@ -0,0 +1,36 @@
# Local JGRPP patches
The `docker/OpenTTD-patches/` directory is an **untracked** clone of
[JGRennison/OpenTTD-patches](https://github.com/JGRennison/OpenTTD-patches) checked out at tag
`jgrpp-0.71.1`. The patches in this directory are the local modifications this project needs on
top of that tag; they are the durable source of truth (the clone itself is not committed).
Current patches:
- `0001-Add-GameScript-API-timetable-getters-to-ScriptOrder.patch` — adds read-only timetable
getters (`GetTimetableWaitTime`, `GetTimetableTravelTime`, `IsWaitTimetabled`,
`IsTravelTimetabled`, `IsWaitFixed`, `IsTravelFixed`, `GetLeaveType`, `GetTimetableMaxSpeed`,
`GetTimetableLateness`, `GetTimetableStartTick`, `GetCurrentOrderTime`,
`GetTimetableTotalDuration`) to the `GSOrder` GameScript class. Required by the AdminBridge
GameScript's `get_timetable` command and the Python client's
`OpenTTDAdminClient.get_timetable()`.
- `0002-Add-GameScript-API-scheduled-dispatch-getters-to-Scr.patch` — adds read-only scheduled
dispatch getters (`GetScheduledDispatchScheduleCount`, `IsScheduledDispatchEnabled`,
`GetScheduledDispatchDuration`, `GetScheduledDispatchStartTick`, `GetScheduledDispatchDelay`,
`GetScheduledDispatchReuseSlots`, `GetScheduledDispatchSlotCount`,
`GetScheduledDispatchSlotOffset`, `GetScheduledDispatchSlotFlags`) to the `GSOrder` GameScript
class. Required by the AdminBridge GameScript's `get_dispatch` command and the Python client's
`OpenTTDAdminClient.get_dispatch()`.
## Applying after a fresh clone
```bash
git clone --branch jgrpp-0.71.1 https://github.com/JGRennison/OpenTTD-patches docker/OpenTTD-patches
git -C docker/OpenTTD-patches am ../patches/*.patch
```
Then build the image as usual (`docker-compose up -d --build` from `docker/`).
If the clone is updated past `jgrpp-0.71.1`, `git am` may conflict — the patches were generated
against that tag and need rebasing in that case.

View File

@@ -31,6 +31,83 @@ The Admin Network has no native packet or `AdminUpdateType` for listing individu
**Important:** the server only forwards `ServerGamescript` packets to admins that have subscribed with `update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)` (enforced server-side in `NetworkAdminGameScript`, which checks `update_frequency[ADMIN_UPDATE_GAMESCRIPT]`). Call `update_frequency()` for `Gamescript` before `list_vehicles()`, or the response is silently dropped.
When a `list_vehicles` request carries a `request_id` field, the AdminBridge GameScript echoes it back in the reply (backward compatible: absent otherwise).
### Timetable Query
The stock GameScript API has no timetable getters, so this project patches the server (see `docker/patches/`) to add read-only getters to `GSOrder` (`GetTimetableWaitTime`, `GetTimetableTravelTime`, `IsWaitTimetabled`, `IsTravelTimetabled`, `IsWaitFixed`, `IsTravelFixed`, `GetLeaveType`, `GetTimetableMaxSpeed`, `GetTimetableLateness`, `GetTimetableStartTick`, `GetCurrentOrderTime`, `GetTimetableTotalDuration`). On top of that, `get_timetable()` sends a request over the same GameScript JSON channel as vehicle listing and awaits the correlated reply — an **authoritative snapshot** of the live game state, unlike the passive observer on the game port (see below).
- **Request:** `{"command": "get_timetable", "vehicle_id": N, "request_id": X}``request_id` is a client-side monotonic counter used to match the reply to the awaiting caller.
- **Reply (success):** `{"command": "get_timetable", "vehicle_id": N, "request_id": X, "lateness": ..., "start_tick": ..., "current_order_time": ..., "total_duration": ..., "orders": [{"position", "wait_time", "travel_time", "wait_timetabled", "travel_timetabled", "wait_fixed", "travel_fixed", "leave_type", "max_speed"}, ...]}` (booleans encoded as 0/1).
- **Reply (error):** same envelope with an `"error"` field instead of the data: `"invalid_vehicle"` (no such vehicle) or `"response_too_large"` (the reply exceeded the admin packet size limit, possible with very many orders). `get_timetable()` raises `ValueError` for these.
Replies carrying a `request_id` that matches a pending request resolve that request and are **not** delivered to the `on_gamescript` callback; all other `ServerGamescript` traffic reaches the callback unchanged. The same `update_frequency` subscription requirement applies (`get_timetable()` subscribes automatically on first use). Since GameScripts do not tick while the game is paused, a query against a paused server times out (`asyncio.TimeoutError`).
### Station Listing
Like vehicles, the Admin Network has no native packet for enumerating individual stations (`ServerCompanyStats` only reports an aggregate per-company station count). `list_stations()` sends a `list_stations` command over the same GameScript JSON channel and the companion AdminBridge GameScript replies with station data through `ServerGamescript` (`{"command": "list_stations", "stations": [{"id", "name", ...}, ...]}`). It is fire-and-forget, so the reply is delivered to the `on_gamescript` callback — subscribe to `Gamescript` updates first, exactly as for `list_vehicles()`. An optional `company_id` field scopes the list to one company.
### Station Query
`get_station()` fetches an authoritative snapshot of one station's live cargo state over the same GameScript JSON channel, awaiting the correlated reply — the station analogue of `get_timetable()`. Unlike timetables, the getters it relies on (`GSStation.GetCargoWaiting`, `GetCargoPlanned`, `GetCargoRating`) are part of the **stock** GameScript API, so this needs no server patch. The per-cargo reply exposes both the **real-time** amount currently waiting and the **planned** amount routed through the station by the cargodist link graph.
- **Request:** `{"command": "get_station", "station_id": N, "request_id": X}``request_id` is the same client-side monotonic counter used by `get_timetable()`, matching the reply to the awaiting caller.
- **Reply (success):** `{"command": "get_station", "station_id": N, "request_id": X, "name": ..., "location": <tile>, "owner": <company_id>, "cargo": [{"cargo_id", "waiting", "planned", "rating"}, ...]}``waiting` is the real-time units at the station (`GetCargoWaiting`), `planned` is the link-graph planned flow (`GetCargoPlanned`, 0 when cargo distribution is off for that cargo), and `rating` is the acceptance rating as a percentage (0-100, `GetCargoRating`) or `null` when the station has no rating for that cargo yet. Only cargo the station has handled appears.
- **Reply (error):** same envelope with an `"error"` field instead of the data: `"invalid_station"` (no such station) or `"response_too_large"`. `get_station()` raises `ValueError` for these.
Correlation, the `update_frequency` subscription requirement (auto-subscribed on first use), and the paused-game timeout behave exactly as described for the Timetable Query above.
### Station Cargo Flow Breakdown
`get_station_cargo()` drills into a single cargo type at one station and returns how its **waiting** (real-time) and **planned** amounts split across the cargo distribution (cargodist) link graph. Cargodist tags every unit with a **source** station (`from`, where it was first loaded) and a **next hop** (`via`, the next station it travels to toward its final destination). There is no per-station store of the *final* destination — the routing destination is the next hop — so the breakdown is offered along those two axes. The GS reads them with the stock `GSStation.GetCargoWaiting{From,Via,FromVia}` / `GetCargoPlanned{From,Via,FromVia}` scalars and the `GSStationList_Cargo{Waiting,Planned}By{From,Via}` (and `…ViaByFrom` / `…FromByVia`) list classes — again no server patch.
- **Request:** `{"command": "get_station_cargo", "station_id": N, "cargo_id": C, "request_id": X}`, optionally with `"from_station"` and/or `"via_station"` filters.
- **Reply (success):** `{"command": "get_station_cargo", "station_id": N, "cargo_id": C, "request_id": X, "waiting": ..., "planned": ..., "waiting_by_from": [{"station", "amount"}, ...], "planned_by_from": [...], "waiting_by_via": [...], "planned_by_via": [...]}`. `waiting`/`planned` are the (filtered) totals; each `*_by_from` list groups by source station and each `*_by_via` list groups by next hop (zero-amount entries omitted). A `station` of `65535` (`STATION_INVALID`) means the source was deleted or — as a next hop — the cargo has no onward routing / is consumed here (also the only next hop for cargo using manual, non-cargodist distribution). Any supplied `from_station`/`via_station` filter is echoed back.
- **Filters:** `via_station` restricts the query (and the `*_by_from` breakdowns) to cargo whose next hop is that station; `from_station` restricts it (and the `*_by_via` breakdowns) to cargo from that source; supplying both makes `waiting`/`planned` the exact source-and-next-hop amount. Pass `65535` to target `STATION_INVALID`.
- **Reply (error):** same envelope with an `"error"` field: `"invalid_station"`, `"invalid_cargo"`, `"invalid_from_station"`/`"invalid_via_station"` (a filter that is neither a valid station nor `STATION_INVALID`), or `"response_too_large"`. `get_station_cargo()` raises `ValueError` for these.
### Dispatch Query
`get_dispatch()` fetches an authoritative snapshot of a vehicle's **scheduled dispatch** state over the GameScript JSON channel, the vehicle analogue of `get_timetable()` for JGRPP's scheduled dispatch feature. Like the timetable getters, the dispatch getters it relies on are added by a **server patch** (`docker/patches/0002-*`, adding `GSOrder.GetScheduledDispatch*` / `IsScheduledDispatchEnabled`), so it needs the patched JGRPP build. Correlation, the auto-subscribe, and the paused-game timeout behave exactly as for the Timetable Query.
- **Request:** `{"command": "get_dispatch", "vehicle_id": N, "request_id": X}`.
- **Reply (success):** `{"command": "get_dispatch", "vehicle_id": N, "request_id": X, "enabled": 0|1, "schedules": [{"index", "duration", "start_tick", "delay", "reuse_slots", "slots": [{"offset", "flags"}, ...]}, ...]}`. `enabled` is whether scheduled dispatch is turned on for the vehicle; each schedule reports its `duration` (ticks), `start_tick`, `delay` (max allowed delay), `reuse_slots` (0/1) and its `slots` (each a departure `offset` within the duration plus a 16-bit `flags` word). These are the same schedules and slots edited by the game-port dispatch methods.
- **Reply (error):** same envelope with an `"error"` field: `"invalid_vehicle"` or `"response_too_large"`. `get_dispatch()` raises `ValueError` for these.
## Vehicle Orders & Timetables (Game Port DoCommands)
Unlike vehicle listing, a vehicle's order list, timetables and scheduled dispatch have no writable GameScript API surface (this project adds read-only timetable and dispatch getters via server patches — see "Timetable Query" and "Dispatch Query" above). Reading and modifying them requires real engine commands (`DoCommand`s) sent over the **game port** (TCP 3979) via `ClientCommand`/`ServerCommand` packets, not the Admin Network. This section covers the wire format; for how to call the methods and what each parameter means, see the [Vehicle Timetables Usage Guide](TIMETABLES.md).
### Command envelope
Both `ClientCommand` and `ServerCommand` share this body: `company (uint8)`, `cmd (uint16 LE, index into the `Commands` enum)`, `error_msg (uint16 LE, StringID, use 0)`, `tile (uint32 LE, always 0 for these commands)`, `payload_len (uint16 LE)`, `payload (payload_len bytes)`, `callback (uint8, use 0)`, `callback_param (uint32 LE, only present if callback != 0)`. `ServerCommand` additionally appends `frame (uint32 LE)` and `my_cmd (uint8 bool)`, and is a **broadcast echo of the request** (no success/failure code) sent to every joined client, not just the sender.
### Payload integer encoding
Command payload fields follow JGRPP's generic serialiser, which picks the wire width from the C++ type's **size**: types of ≤1 byte are sent as a fixed `uint8`, exactly 2 bytes as a fixed `uint16` (LE), and 4/8-byte types as a variable-length **varuint** (`write_varuint`/`read_varuint` in `protocol.py` — a UTF-8-like prefix encoding, not LEB128; signed fields use zigzag via `write_varuint_signed`/`read_varuint_signed`). This is why `VehicleID` (a 4-byte pool id) is a varuint while `VehicleOrderID` (a `uint16`) is a fixed `uint16`.
### Command IDs and payload tuples
| Method | `cmd` | payload |
|---|---|---|
| `add_order()` | 52 (`InsertOrder`) | `VehicleID (varuint), sel_ord (uint16), order_type (uint8), order_flags (uint16), DestinationID (uint16)` |
| `remove_order()` | 51 (`DeleteOrder`) | `VehicleID (varuint), VehicleOrderID (uint16)` |
| `change_timetable()` | 174 (`ChangeTimetable`) | `VehicleID (varuint), VehicleOrderID (uint16), ModifyTimetableFlags (uint8), value (varuint), ModifyTimetableCtrlFlags (uint8)` |
| `set_vehicle_on_time()` | 176 (`SetVehicleOnTime`) | `VehicleID (varuint), apply_to_group (uint8 bool)` |
| `autofill_timetable()` | 177 (`AutofillTimetable`) | `VehicleID (varuint), bool (uint8), bool (uint8)` |
| `set_timetable_start()` | 180 (`SetTimetableStart`) | `VehicleID (varuint), bool (uint8), StateTicks (signed varuint)` |
| `set_scheduled_dispatch()` | 205 (`SchDispatch`) | `VehicleID (varuint), enabled (uint8 bool)` |
| `add_dispatch_slot()` | 206 (`SchDispatchAdd`) | `VehicleID (varuint), schedule_index (varuint), offset (varuint), interval (varuint), extra_slots (varuint), slot_flags (uint16), route_id (uint8)` |
| `remove_dispatch_slot()` | 207 (`SchDispatchRemove`) | `VehicleID (varuint), schedule_index (varuint), offset (varuint)` |
| `set_dispatch_duration()` | 208 (`SchDispatchSetDuration`) | `VehicleID (varuint), schedule_index (varuint), duration (varuint)` |
| `set_dispatch_start_date()` | 209 (`SchDispatchSetStartDate`) | `VehicleID (varuint), schedule_index (varuint), StateTicks (signed varuint)` |
| `clear_dispatch_schedule()` | 213 (`SchDispatchClear`) | `VehicleID (varuint), schedule_index (varuint)` |
| `add_dispatch_schedule()` | 214 (`SchDispatchAddNewSchedule`) | `VehicleID (varuint), StateTicks (signed varuint), duration (varuint)` |
| `remove_dispatch_schedule()` | 215 (`SchDispatchRemoveSchedule`) | `VehicleID (varuint), schedule_index (varuint)` |
### Scheduled dispatch (JGRPP)
A vehicle's order list can carry several **dispatch schedules**, each with a duration, a start tick and a set of departure **slots** (offsets within the duration). The methods above edit them over the game port (`add_dispatch_schedule()`/`remove_dispatch_schedule()` create and delete schedules; `add_dispatch_slot()`/`remove_dispatch_slot()`/`clear_dispatch_schedule()` manage a schedule's slots; `set_dispatch_duration()`/`set_dispatch_start_date()` adjust a schedule; `set_scheduled_dispatch()` toggles the feature for the vehicle). `add_dispatch_slot()` can add several evenly spaced slots at once via its `interval`/`extra_slots` parameters. The stock JGRPP command set covers ~22 dispatch commands (routes, departure tags, per-slot flags, adjust/swap/duplicate, …); the client implements this common core. There is no game-port read; for an authoritative view of the resulting schedules use the Admin Network's `get_dispatch()` (see "Dispatch Query" below).
### Adding & removing orders
`add_order()` issues `CMD_INSERT_ORDER`, which appends a new order before `sel_ord` (pass `0xFFFF`/`INVALID_VEH_ORDER_ID` to append to the end). The client currently builds "go to station" orders only. The `order_type` byte is bit-packed: **bits 0-3** hold the `OrderType` (`1` = `OT_GOTO_STATION`), **bits 4-5** the `OrderStopLocation`, and **bits 6-7** the `OrderNonStopFlags`. The stop location defaults to `PlatformFarEnd` (`2`) because near-end/middle/through are **train-only** and the server rejects (`CMD_ERROR`, no state change) any other value for road vehicles, ships, or aircraft. `order_flags` is the 16-bit load/unload word (`0` = load-if-possible + unload-if-possible). `DestinationID` is the target `StationID`. `remove_order()` issues `CMD_DELETE_ORDER` for the order at a given position. Both are **broadcast** back as `ServerCommand` like any DoCommand; the client does not currently decode those echoes into observed order state, so verify results via the admin `get_timetable()` order count.
### Ownership requirement
A command is rejected unless it's issued by the company that owns the target vehicle — join that company via `join_company()` with a real company id (not 255/spectator) before calling any order or timetable method. A malformed or wrong-company packet is treated as illegal and the client is kicked; a well-formed command that merely fails validation (e.g. an order the vehicle can't serve) is silently dropped with no state change and no kick.
### Reading timetables — no query command exists on the game port
There is no getter `DoCommand` for orders/timetables anywhere in the protocol. `get_vehicle_timetable()` works by passively decoding `ServerCommand` broadcasts (including the sender's own) as they arrive — it only reflects **changes made after the client joined**. A vehicle's pre-existing timetable (set before this client connected) is invisible until something changes it again; seeing it upfront would require parsing the `ORDR`/`VEHS` chunks of the initial savegame transfer (`ServerMapData`), which this client does not implement. For an authoritative read, use the Admin Network's `get_timetable()` instead (see "Timetable Query" above).
## Stream Encryption (AEAD)
Once `ServerEnableEncryption` is received, all subsequent packets use **XChaCha20-Poly1305** (Authenticated Encryption with Associated Data).

409
docs/TIMETABLES.md Normal file
View File

@@ -0,0 +1,409 @@
# Vehicle Timetables: Usage Guide
This guide covers the timetable API: writing via `OpenTTDClient`'s `change_timetable()`,
`autofill_timetable()`, `set_timetable_start()`, `set_vehicle_on_time()`, and reading via
`OpenTTDAdminClient.get_timetable()` (authoritative, recommended) or `OpenTTDClient`'s
`get_vehicle_timetable()` (passive change observer). For wire-format internals (packet layout,
varuint encoding, command IDs), see [PROTOCOL.md](PROTOCOL.md#vehicle-timetables-game-port-docommands).
This guide is about *how to call these methods and what their parameters mean*, with worked examples.
## Reading: `get_timetable()` — authoritative snapshot (recommended)
An **awaitable** method on `OpenTTDAdminClient` (admin port, TCP 3977) that queries the real
timetable state from the running game via the AdminBridge GameScript. Unlike the observer
approach below, it works for timetables set **before** you connected and returns the game's
**actual** state, not the last requested change. No `join_company` needed — it's read-only and
sees every company's vehicles.
```python
import asyncio
from openttd import OpenTTDAdminClient
async def main():
admin = OpenTTDAdminClient("127.0.0.1", admin_name="TimetableReader")
await admin.connect(admin_password="asd")
await admin.joined.wait()
data = await admin.get_timetable(7)
print(data)
await admin.quit()
asyncio.run(main())
```
Returns a `dict` shaped like:
```python
{
"command": "get_timetable",
"vehicle_id": 7,
"lateness": 0, # ticks late; negative = running early
"start_tick": 1000000, # absolute StateTicks the timetable starts at; 0 = not started
"current_order_time": 42, # ticks spent on the current order so far
"total_duration": 5400, # full timetable round-trip in ticks; -1 = timetable incomplete
"orders": [
{
"position": 0,
"wait_time": 120, # ticks (raw stored value)
"travel_time": 300, # ticks (raw stored value)
"wait_timetabled": 1, # 1 = wait time explicitly set, 0 = not timetabled
"travel_timetabled": 1,
"wait_fixed": 0, # 1 = locked against autofill
"travel_fixed": 0,
"leave_type": 0, # 0 normal, 1 leave early, 2 early if any cargo full, 3 early if all full
"max_speed": 65535, # order speed cap; 65535 = no cap
},
# ... one entry per order position
],
}
```
Errors: raises `ValueError` when the GameScript reports one (`invalid_vehicle` for a nonexistent
vehicle id; `response_too_large` if a very long order list overflows the admin packet limit),
`asyncio.TimeoutError` when no reply arrives within `timeout` (default 5.0s — note GameScripts
don't run while the game is **paused**, so a paused server always times out), and
`ConnectionError` if the admin connection drops mid-query.
Requirements: the server must run the bundled AdminBridge GameScript **and** the patched JGRPP
build with the `GSOrder` timetable getters (both included in this repo's `docker/` setup — see
`docker/patches/README.md`). The Gamescript update-frequency subscription it needs is set up
automatically on first call.
## Writing (and the legacy observer): things you must know
1. **You must join the vehicle's own company to write.** The `change_timetable()` family are real
game commands (`DoCommand`s) on the game port, not admin-network calls. Join with
`client.join_company(company_id=<id>, company_password=...)` using the id of the company that
owns the vehicle — spectators (`company_id=255`, the default) are rejected. Sending a command
for a vehicle you don't own also fails.
2. **`get_vehicle_timetable()` is a passive observer, not a query.** It watches the
`ServerCommand` broadcasts the server sends to every joined client whenever *anyone* changes a
timetable. Use it for live change monitoring on the game port; prefer `get_timetable()` above
for reading actual state. Its limitations:
- It only reflects changes made **after your client joined**. A vehicle's pre-existing
timetable (set before you connected) is invisible until something changes it again.
- It reflects what was **requested**, not a confirmed result — the wire protocol has no
success/failure code, so a command that the server silently rejects (wrong owner, invalid
order, etc.) still updates your local view as if it succeeded.
## Quick start
```python
import asyncio
from openttd import OpenTTDClient
from openttd.protocol import ModifyTimetableFlags
async def main():
client = OpenTTDClient(host="127.0.0.1", username="TimetableBot")
await client.connect(server_password="asd")
# Must be a real company you own vehicles in -- not 255 (spectator).
await client.join_company(company_id=0, company_password="")
await client.joined.wait()
# Set order 0's wait time to 120 ticks for vehicle 7.
await client.change_timetable(7, 0, ModifyTimetableFlags.WaitTime, 120)
# Give the broadcast a moment to round-trip back to us.
await asyncio.sleep(1.0)
print(client.get_vehicle_timetable(7))
# -> {'orders': {0: {'wait_time': 120}}}
await client.quit()
asyncio.run(main())
```
## `change_timetable(vehicle_id, order_position, flag, value, clear_field=False)`
Changes one field of one order's timetable entry. This is the general-purpose "edit a cell in the
Timetable window" command — every other kind of edit (wait time, travel time, max speed, fixed
flags, leave type, dispatch schedule assignment) goes through this one method, distinguished by
`flag`.
| Parameter | Type | Meaning |
|---|---|---|
| `vehicle_id` | `int` | The `VehicleID` whose order list you're editing. You must own the company this vehicle belongs to. |
| `order_position` | `int` | Zero-based index into the vehicle's order list (order 0, order 1, ...). Must be a valid, existing order — you can't create orders with this method, only edit existing ones. |
| `flag` | `ModifyTimetableFlags` | Which field of the order to change (see table below). Import from `openttd.protocol`. |
| `value` | `int` | The new value. **Its meaning depends entirely on `flag`** — see below. |
| `clear_field` | `bool` | Only meaningful when `flag` is `WaitTime` or `TravelTime`. See "Clearing a field" below. Default `False`. |
### `ModifyTimetableFlags` values and what `value` means for each
| Flag | What it changes | `value` meaning |
|---|---|---|
| `ModifyTimetableFlags.WaitTime` | How long the vehicle waits at this order (e.g. at a station) | Wait time **in game ticks** |
| `ModifyTimetableFlags.TravelTime` | How long the vehicle takes to travel to this order | Travel time **in game ticks** |
| `ModifyTimetableFlags.TravelSpeed` | The order's max speed cap | Max speed in the order's internal speed unit (the same number shown in the Timetable window's speed column). Pass `0` to **remove** the speed cap entirely (no clamp) |
| `ModifyTimetableFlags.SetWaitFixed` | Whether the wait time is "fixed" (locked, so autofill won't overwrite it) | `1` to fix, `0` to unfix |
| `ModifyTimetableFlags.SetTravelFixed` | Whether the travel time is "fixed" (locked) | `1` to fix, `0` to unfix |
| `ModifyTimetableFlags.SetLeaveType` | When the vehicle is allowed to leave this order early | `0` = normal (leave when timetabled), `1` = leave as soon as possible, `2` = leave early if any cargo is fully loaded, `3` = leave early if all cargo is fully loaded |
| `ModifyTimetableFlags.AssignSchedule` | Which scheduled-dispatch schedule this order is tied to | A schedule index (`0`, `1`, ...), or `0xFFFFFFFF` (4294967295) to unassign (no schedule) |
A "tick" is the game's base simulation unit; how much real time it represents depends on the
server's day-length setting, so there's no fixed ticks-per-second conversion you can rely on
across servers.
### Clearing a field
`clear_field=True` only makes sense with `flag=WaitTime` or `flag=TravelTime`, and **you must also
pass `value=0`** — the server rejects the command (silently, as always — you'll only notice because
`get_vehicle_timetable()` won't show the change you expected) if `clear_field=True` and `value != 0`.
Clearing is different from just setting the time to `0`:
- `change_timetable(v, 0, ModifyTimetableFlags.WaitTime, 0)` — sets wait time to exactly 0 ticks,
but the order is still considered "timetabled" (has an explicit time).
- `change_timetable(v, 0, ModifyTimetableFlags.WaitTime, 0, clear_field=True)` — removes the
timetabled wait time entirely (back to "no time set").
### Examples
```python
from openttd.protocol import ModifyTimetableFlags
# Set order 0's wait time to 120 ticks.
await client.change_timetable(7, 0, ModifyTimetableFlags.WaitTime, 120)
# Set order 1's travel time to 300 ticks.
await client.change_timetable(7, 1, ModifyTimetableFlags.TravelTime, 300)
# Cap order 0's speed at 80 (speed units), then remove the cap again.
await client.change_timetable(7, 0, ModifyTimetableFlags.TravelSpeed, 80)
await client.change_timetable(7, 0, ModifyTimetableFlags.TravelSpeed, 0) # 0 = no cap
# Lock order 0's wait time so autofill won't touch it.
await client.change_timetable(7, 0, ModifyTimetableFlags.SetWaitFixed, 1)
# Let the vehicle leave order 2 as soon as it's loaded, instead of waiting for the timetabled time.
await client.change_timetable(7, 2, ModifyTimetableFlags.SetLeaveType, 1) # OLT_LEAVE_EARLY
# Assign order 0 to scheduled-dispatch schedule 0, then unassign it.
await client.change_timetable(7, 0, ModifyTimetableFlags.AssignSchedule, 0)
await client.change_timetable(7, 0, ModifyTimetableFlags.AssignSchedule, 0xFFFFFFFF)
# Clear order 0's wait time back to "not timetabled".
await client.change_timetable(7, 0, ModifyTimetableFlags.WaitTime, 0, clear_field=True)
```
## `autofill_timetable(vehicle_id, autofill=True, preserve_wait_time=False)`
Turns the "Autofill timetable" feature on or off for a vehicle. While autofill is active, the game
fills in wait/travel times automatically as the vehicle completes each order, instead of you
setting them manually with `change_timetable()`.
| Parameter | Type | Meaning |
|---|---|---|
| `vehicle_id` | `int` | The vehicle to enable/disable autofill for. |
| `autofill` | `bool` | `True` to start autofilling (also clears the "timetable has started" state — enabling autofill is how you (re)start building a timetable from scratch). `False` to stop. Default `True`. |
| `preserve_wait_time` | `bool` | Only relevant when `autofill=True`. If `True`, autofill only *increases* existing wait times, never shortens them, instead of overwriting them outright. Default `False`. |
### Examples
```python
# Start autofilling vehicle 7's timetable from scratch.
await client.autofill_timetable(7, autofill=True, preserve_wait_time=False)
# Start autofilling, but never shrink wait times the vehicle already has set.
await client.autofill_timetable(7, autofill=True, preserve_wait_time=True)
# Stop autofilling once you're happy with the result.
await client.autofill_timetable(7, autofill=False)
```
## `set_timetable_start(vehicle_id, timetable_all, start_date)`
Sets when a vehicle's timetable begins running.
| Parameter | Type | Meaning |
|---|---|---|
| `vehicle_id` | `int` | The vehicle whose timetable start to set. |
| `timetable_all` | `bool` | `True` to apply this start date to every vehicle that shares this vehicle's order list (a "vehicle group" running the same route); `False` to affect only this one vehicle. |
| `start_date` | `int` | An **absolute `StateTicks` value** — OpenTTD's internal tick counter that always advances at the same rate regardless of day-length settings. It is *not* a calendar date and *not* relative to "now". |
**About `start_date`:** this library doesn't currently expose "what is the current `StateTicks`
value" anywhere (the admin `ServerDate` packet reports a calendar date, which is a different,
day-length-dependent counter). In practice you'll usually either: read a `timetable_start` value
already observed via `get_vehicle_timetable()` on another vehicle in the same group and reuse it,
or coordinate the value out-of-band (e.g. from an in-game GameScript, or a known baseline) rather
than computing "now" purely from this client.
### Examples
```python
# Start vehicle 7's own timetable at StateTicks 1_000_000.
await client.set_timetable_start(7, timetable_all=False, start_date=1_000_000)
# Start the timetable for every vehicle sharing vehicle 7's orders, all at the same tick.
await client.set_timetable_start(7, timetable_all=True, start_date=1_000_000)
```
## `set_vehicle_on_time(vehicle_id, apply_to_group=False)`
Resets a vehicle's **lateness counter to zero** (marks it on-time). This command can only reduce
lateness to zero — there is no way to use it to mark a vehicle as *late*.
| Parameter | Type | Meaning |
|---|---|---|
| `vehicle_id` | `int` | The vehicle to reset lateness for. |
| `apply_to_group` | `bool` | `False` (default): reset only this vehicle. `True`: reset lateness for every vehicle sharing this vehicle's order list, by the same amount (so their relative spacing is preserved), instead of just this one. |
Note: if `apply_to_group=False` and the vehicle's timetable hasn't been started yet (see
`set_timetable_start()`), the server rejects the command — but since there's no success/failure
signal on the wire, you won't see an error, `get_vehicle_timetable()` will just show the request
was made without the underlying lateness actually having changed.
### Examples
```python
# Reset lateness for just this vehicle.
await client.set_vehicle_on_time(7, apply_to_group=False)
# Reset lateness for the whole group of vehicles sharing vehicle 7's orders.
await client.set_vehicle_on_time(7, apply_to_group=True)
```
## `get_vehicle_timetable(vehicle_id)`
A **synchronous** method (no `await`, no network round-trip) that returns whatever this client has
locally observed about a vehicle's timetable so far, or `None` if nothing has been observed for
that vehicle id yet.
```python
entry = client.get_vehicle_timetable(7)
```
Returns either `None`, or a `dict` shaped like:
```python
{
"orders": {
0: {"wait_time": 120, "wait_time_fixed": True},
2: {"travel_time": 300, "leave_type": 1},
# only order positions that have been touched by an observed change_timetable() appear here
},
"autofill": True, # present after an observed autofill_timetable()
"autofill_preserve_wait_time": False,
"timetable_start": 1000000, # present after an observed set_timetable_start()
"timetable_all": False,
"on_time_apply_to_group": False, # present after an observed set_vehicle_on_time()
}
```
Every top-level key is optional and only appears once the corresponding change has actually been
observed — a freshly-joined client that hasn't seen any broadcasts yet for a vehicle returns `None`
for it, and a vehicle that's only had its wait time changed won't have an `"autofill"` key at all.
Per-order fields inside `"orders"` follow the same rule: only fields that have been explicitly set
via `change_timetable()` appear; a cleared field (`clear_field=True`) is stored as `None` rather
than being removed, so you can distinguish "never touched" (key absent) from "explicitly cleared"
(key present, value `None`).
## Putting it together
```python
import asyncio
from openttd import OpenTTDClient
from openttd.protocol import ModifyTimetableFlags
async def build_timetable(client, vehicle_id):
# 1. Let autofill do a first pass, preserving anything already set.
await client.autofill_timetable(vehicle_id, autofill=True, preserve_wait_time=True)
await asyncio.sleep(1.0)
# 2. Manually lock in the wait time for a specific order once you're happy with it.
await client.change_timetable(vehicle_id, 0, ModifyTimetableFlags.WaitTime, 90)
await client.change_timetable(vehicle_id, 0, ModifyTimetableFlags.SetWaitFixed, 1)
await asyncio.sleep(1.0)
# 3. Turn autofill off and start the timetable running for the whole group.
await client.autofill_timetable(vehicle_id, autofill=False)
await client.set_timetable_start(vehicle_id, timetable_all=True, start_date=1_000_000)
await asyncio.sleep(1.0)
print(client.get_vehicle_timetable(vehicle_id))
async def main():
client = OpenTTDClient(host="127.0.0.1", username="TimetableBot")
await client.connect(server_password="asd")
await client.join_company(company_id=0, company_password="")
await client.joined.wait()
await build_timetable(client, vehicle_id=7)
await client.quit()
asyncio.run(main())
```
## Adding and removing orders
Beyond editing an order's timetable fields, you can change the order list itself. Both commands go
over the game port and require being joined to the company that owns the vehicle (like the timetable
methods above).
```python
# Append a "go to station" order (station id 6) to the end of vehicle 7's order list.
await client.add_order(7, 6)
# Insert one before position 0 instead of appending.
await client.add_order(7, 6, before_position=0)
# Non-stop / stop-location can be customised (defaults suit every vehicle type).
from openttd.protocol import OrderNonStopFlags
await client.add_order(7, 6, non_stop=OrderNonStopFlags.NoStopAtIntermediate)
# Delete the order at a given position.
await client.remove_order(7, 0)
```
`add_order()` builds "go to station" orders. `stop_location` defaults to `PlatformFarEnd` because the
other stop locations are train-only and rejected for road vehicles, ships and aircraft. There is no
game-port query for the resulting order list; confirm changes with the admin `get_timetable()` order
count (see [PROTOCOL.md](PROTOCOL.md#adding--removing-orders)).
## Scheduled dispatch (JGRPP)
Scheduled dispatch lets a vehicle depart on a fixed schedule of slots rather than purely by
timetable. A vehicle's order list can hold several dispatch schedules, each with a duration, a start
tick and a set of departure slots. The edit commands go over the game port and require being joined
to the owning company; the authoritative read is on the admin client.
```python
# Create a schedule (start tick 0, duration 3000 ticks) — it becomes the next schedule index.
await client.add_dispatch_schedule(7, 0, 3000)
# Add departure slots at offsets 500 and 1500 within schedule 0's duration.
await client.add_dispatch_slot(7, 0, 500)
await client.add_dispatch_slot(7, 0, 1500)
# Add several evenly spaced slots at once: offset 0, then +250 three more times.
await client.add_dispatch_slot(7, 0, 0, interval=250, extra_slots=3)
# Adjust the schedule, then turn scheduled dispatch on for the vehicle.
await client.set_dispatch_duration(7, 0, 4000)
await client.set_dispatch_start_date(7, 0, 1_000_000)
await client.set_scheduled_dispatch(7, True)
# Remove a slot, clear a schedule's slots, or remove the whole schedule.
await client.remove_dispatch_slot(7, 0, 1500)
await client.clear_dispatch_schedule(7, 0)
await client.remove_dispatch_schedule(7, 0)
```
Read the live state back over the admin connection (requires the patched JGRPP build, see
[docker/patches/README.md](../docker/patches/README.md)):
```python
data = await admin.get_dispatch(7)
# {"enabled": 1, "schedules": [{"index": 0, "duration": 4000, "start_tick": 1000000,
# "delay": 0, "reuse_slots": 0, "slots": [{"offset": 500, "flags": 0}, ...]}]}
```
The client implements a common core of the ~22 JGRPP dispatch commands; advanced operations
(departure routes/tags, per-slot flags, adjust/swap/duplicate) are not wrapped yet.
## See also
- [PROTOCOL.md — Vehicle Orders & Timetables](PROTOCOL.md#vehicle-orders--timetables-game-port-docommands) for the underlying wire format.
- [PROTOCOL.md — Dispatch Query](PROTOCOL.md#dispatch-query) for the admin `get_dispatch()` read.
- [ARCHITECTURE.md](ARCHITECTURE.md) for how `OpenTTDClient` fits into the rest of the library.

View File

@@ -5,7 +5,13 @@ 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 .protocol import PacketGameType, OpenTTDProtocol, PacketAdminType, OpenTTDAdminProtocol, NetworkAuthenticationMethod
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:
@@ -22,6 +28,7 @@ class OpenTTDClient:
self.joined = asyncio.Event()
self.shutdown_event = asyncio.Event()
self.client_id = None
self.vehicle_timetables = {}
# Internal crypto
self._server_password = ""
@@ -62,6 +69,186 @@ class OpenTTDClient:
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.")
@@ -199,7 +386,55 @@ class OpenTTDClient:
async def receive_ServerMapData(self, source, **kwargs): pass
async def receive_ServerConfigurationUpdate(self, source, **kwargs): pass
async def receive_ServerExternalChat(self, source, **kwargs): pass
async def receive_ServerCommand(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
@@ -231,6 +466,11 @@ class OpenTTDAdminClient:
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
@@ -260,6 +500,10 @@ class OpenTTDAdminClient:
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):
@@ -329,6 +573,154 @@ class OpenTTDAdminClient:
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
@@ -424,10 +816,17 @@ class OpenTTDAdminClient:
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(kwargs.get('data'))
self.on_gamescript(data)
else:
self.log.info(f"GAMESCRIPT: {kwargs.get('data')}")
self.log.info(f"GAMESCRIPT: {data}")
async def receive_ServerDate(self, source, **kwargs): pass
async def receive_ServerFull(self, source, **kwargs): await self.quit()

View File

@@ -5,6 +5,101 @@ 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):
DeleteOrder = 51
InsertOrder = 52
ChangeTimetable = 174
SetVehicleOnTime = 176
AutofillTimetable = 177
SetTimetableStart = 180
# Scheduled dispatch (JGRPP)
SchDispatch = 205
SchDispatchAdd = 206
SchDispatchRemove = 207
SchDispatchSetDuration = 208
SchDispatchSetStartDate = 209
SchDispatchClear = 213
SchDispatchAddNewSchedule = 214
SchDispatchRemoveSchedule = 215
# Sentinel VehicleOrderID meaning "append to the end of the order list" for InsertOrder.
INVALID_VEH_ORDER_ID = 0xFFFF
class OrderType(IntEnum):
"""OrderType occupies bits 0-3 of an order's `type` byte (bits 6-7 hold OrderNonStopFlags)."""
GotoStation = 1
GotoDepot = 2
GotoWaypoint = 6
class OrderNonStopFlags(IntEnum):
"""Packed into bits 6-7 of an order's `type` byte."""
StopEverywhere = 0
NoStopAtIntermediate = 1
NoStopAtDestination = 2
NoStopAtAny = 3
class OrderStopLocation(IntEnum):
"""Packed into bits 4-5 of an order's `type` byte. Near-end/middle/through are train-only;
FarEnd is the only value the server accepts for every vehicle type, so it is the safe default."""
PlatformNearEnd = 0
PlatformMiddle = 1
PlatformFarEnd = 2
PlatformThrough = 3
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 +324,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

124
main.py
View File

@@ -7,6 +7,7 @@ import os
sys.path.append(os.path.join(os.path.dirname(__file__), 'lib'))
from openttd import OpenTTDClient
from openttd.protocol import ModifyTimetableFlags
# Configuration
SERVER_HOST = "127.0.0.1"
@@ -17,6 +18,123 @@ SERVER_PASSWORD = "asd"
COMPANY_ID = 0 # "Én transport"
COMPANY_PASSWORD = "asd123"
# A vehicle owned by COMPANY_ID, used to demonstrate timetable get/set below. Set to a real
# vehicle id to see it in action; leave as None to skip the demonstration.
DEMO_VEHICLE_ID = 7
# A station DEMO_VEHICLE_ID can legally serve, used to demonstrate add_order/remove_order.
DEMO_STATION_ID = 6
async def demo_timetable_workflow(client, vehicle_id):
"""A deliberately thorough walk-through of the timetable API: every ModifyTimetableFlags
variant, clear_field, autofill, timetable start, and lateness reset, on a vehicle assumed
to have at least two orders (positions 0 and 1)."""
ORDER_A, ORDER_B = 0, 1
def show(label):
print(f" -> [{label}] {client.get_vehicle_timetable(vehicle_id)}")
print(f"=== Timetable demo starting for vehicle {vehicle_id} ===")
# 1. Wait/travel times, in game ticks.
print("--- Step 1: set wait/travel times ---")
await client.change_timetable(vehicle_id, ORDER_A, ModifyTimetableFlags.WaitTime, 90)
await client.change_timetable(vehicle_id, ORDER_A, ModifyTimetableFlags.TravelTime, 240)
await client.change_timetable(vehicle_id, ORDER_B, ModifyTimetableFlags.WaitTime, 45)
await asyncio.sleep(0.5)
show("wait/travel times set")
# 2. Lock order A's wait time so autofill won't overwrite it later.
print("--- Step 2: fix order A's wait time ---")
await client.change_timetable(vehicle_id, ORDER_A, ModifyTimetableFlags.SetWaitFixed, 1)
await asyncio.sleep(0.5)
show("order A wait time fixed")
# 3. Cap order B's speed, then remove the cap again (0 = uncapped).
print("--- Step 3: cap and uncap order B's speed ---")
await client.change_timetable(vehicle_id, ORDER_B, ModifyTimetableFlags.TravelSpeed, 80)
await asyncio.sleep(0.5)
show("order B speed capped at 80")
await client.change_timetable(vehicle_id, ORDER_B, ModifyTimetableFlags.TravelSpeed, 0)
await asyncio.sleep(0.5)
show("order B speed cap removed")
# 4. Let the vehicle leave order B early once any cargo is fully loaded.
print("--- Step 4: change order B's leave type ---")
await client.change_timetable(vehicle_id, ORDER_B, ModifyTimetableFlags.SetLeaveType, 2)
await asyncio.sleep(0.5)
show("order B leave type: leave early if any cargo full")
# 5. Assign order A to scheduled-dispatch schedule 0, then unassign it again.
print("--- Step 5: assign and unassign a dispatch schedule ---")
await client.change_timetable(vehicle_id, ORDER_A, ModifyTimetableFlags.AssignSchedule, 0)
await asyncio.sleep(0.5)
show("order A assigned to dispatch schedule 0")
await client.change_timetable(vehicle_id, ORDER_A, ModifyTimetableFlags.AssignSchedule, 0xFFFFFFFF)
await asyncio.sleep(0.5)
show("order A unassigned from dispatch schedule")
# 6. Clear order B's wait time entirely (distinct from setting it to 0).
print("--- Step 6: clear order B's wait time ---")
await client.change_timetable(vehicle_id, ORDER_B, ModifyTimetableFlags.WaitTime, 0, clear_field=True)
await asyncio.sleep(0.5)
show("order B wait time cleared")
# 7. Autofill: start it preserving existing (fixed) wait times, then turn it off again.
print("--- Step 7: toggle autofill ---")
await client.autofill_timetable(vehicle_id, autofill=True, preserve_wait_time=True)
await asyncio.sleep(0.5)
show("autofill enabled (preserving wait times)")
await client.autofill_timetable(vehicle_id, autofill=False)
await asyncio.sleep(0.5)
show("autofill disabled")
# 8. Start the timetable for this vehicle only, then restart it for the whole group.
print("--- Step 8: set timetable start ---")
await client.set_timetable_start(vehicle_id, timetable_all=False, start_date=1_000_000)
await asyncio.sleep(0.5)
show("timetable started (this vehicle only)")
await client.set_timetable_start(vehicle_id, timetable_all=True, start_date=1_500_000)
await asyncio.sleep(0.5)
show("timetable restarted (whole group)")
# 9. Reset lateness for this vehicle, then for the whole group sharing its orders.
print("--- Step 9: reset lateness ---")
await client.set_vehicle_on_time(vehicle_id, apply_to_group=False)
await asyncio.sleep(0.5)
show("lateness reset (this vehicle only)")
await client.set_vehicle_on_time(vehicle_id, apply_to_group=True)
await asyncio.sleep(0.5)
show("lateness reset (whole group)")
# 10. Add an order to the front of the list, then remove it again (net-zero, so the
# vehicle's route is left unchanged). Inserting before position 0 and deleting
# position 0 needs no knowledge of the existing order count.
print(f"--- Step 10: add then remove a 'go to station {DEMO_STATION_ID}' order ---")
await client.add_order(vehicle_id, DEMO_STATION_ID, before_position=0)
await asyncio.sleep(0.5)
print(" -> inserted a goto-station order at position 0")
await client.remove_order(vehicle_id, 0)
await asyncio.sleep(0.5)
print(" -> removed it again (route restored)")
# 11. Scheduled dispatch: create a schedule with two departure slots, enable it, then tear it
# all down again so the vehicle is left as it started. Read it back with
# OpenTTDAdminClient.get_dispatch() (see main_admin.py); the game port has no dispatch read.
print("--- Step 11: scheduled dispatch create/enable, then clean up ---")
await client.add_dispatch_schedule(vehicle_id, start_tick=0, duration=3000)
await client.add_dispatch_slot(vehicle_id, 0, 500)
await client.add_dispatch_slot(vehicle_id, 0, 1500)
await client.set_scheduled_dispatch(vehicle_id, True)
await asyncio.sleep(0.5)
print(" -> created schedule 0 with 2 slots and enabled scheduled dispatch")
await client.set_scheduled_dispatch(vehicle_id, False)
await client.remove_dispatch_schedule(vehicle_id, 0)
await asyncio.sleep(0.5)
print(" -> disabled and removed the schedule (restored)")
print(f"=== Timetable demo finished. Final state for vehicle {vehicle_id}: ===")
print(f" {client.get_vehicle_timetable(vehicle_id)}")
async def run_client():
# 1. Initialize high-level client
username = sys.argv[1] if len(sys.argv) > 1 else "Modular_Joiner"
@@ -41,7 +159,11 @@ async def run_client():
await client.joined.wait()
print(f"--- Successfully joined! Client ID: {client.client_id} ---")
# 6. Lifecycle management
# 6. Timetable demonstration (requires DEMO_VEHICLE_ID to be owned by COMPANY_ID)
if DEMO_VEHICLE_ID is not None:
await demo_timetable_workflow(client, DEMO_VEHICLE_ID)
# 7. Lifecycle management
# We wait for either a manual shutdown signal or a 10s timeout
try:
await asyncio.wait_for(client.shutdown_event.wait(), timeout=10.0)

View File

@@ -50,6 +50,41 @@ async def run_admin():
print("--- Requesting vehicle info via GameScript ---")
await admin.list_vehicles()
# Capture station-list replies (delivered to on_gamescript, like list_vehicles) while
# still logging every other GameScript message.
stations = []
def gamescript_capture(data):
if isinstance(data, dict) and "stations" in data:
stations.append(data["stations"])
gamescript_logger(data)
admin.on_gamescript = gamescript_capture
print("--- Requesting station info via GameScript ---")
await admin.list_stations()
await asyncio.sleep(1)
# Fetch one station's authoritative live cargo (real-time waiting + planned).
if stations and stations[-1]:
sid = stations[-1][0]["id"]
try:
data = await admin.get_station(sid, timeout=10.0)
print(f"--- Station {sid} ({data.get('name')}) cargo: real-time waiting vs planned ---")
for cargo in data.get("cargo", []):
print(f" cargo {cargo['cargo_id']}: waiting={cargo['waiting']} "
f"planned={cargo['planned']} rating={cargo['rating']}")
# Break the first cargo down by source station and by next hop (routing destination).
if data.get("cargo"):
cid = data["cargo"][0]["cargo_id"]
flow = await admin.get_station_cargo(sid, cid, timeout=10.0)
print(f"--- Station {sid} cargo {cid} flow breakdown (station 65535 = none/deleted) ---")
print(f" waiting by source: {flow['waiting_by_from']}")
print(f" waiting by next hop: {flow['waiting_by_via']}")
print(f" planned by source: {flow['planned_by_from']}")
print(f" planned by next hop: {flow['planned_by_via']}")
except Exception as e:
print(f"!!! station query failed: {e}")
await asyncio.sleep(5)
print("--- Quitting ---")
await admin.quit()

View File

@@ -46,6 +46,20 @@ async def test_admin_list_vehicles():
assert decode_gamescript_payload(proto.sent[0]) == {"command": "list_vehicles"}
assert decode_gamescript_payload(proto.sent[1]) == {"command": "list_vehicles", "company_id": 2}
@pytest.mark.asyncio
async def test_admin_list_stations():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
proto = MockProtocol()
client._protocol = proto
client._transport = MockTransport()
await client.list_stations()
await client.list_stations(company_id=2)
assert len(proto.sent) == 2
assert decode_gamescript_payload(proto.sent[0]) == {"command": "list_stations"}
assert decode_gamescript_payload(proto.sent[1]) == {"command": "list_stations", "company_id": 2}
@pytest.mark.asyncio
async def test_admin_client_connect_and_actions(monkeypatch):
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
@@ -187,3 +201,240 @@ async def test_admin_client_connect_and_actions(monkeypatch):
client._protocol = BadProtocol()
await client.quit()
assert client.shutdown_event.is_set()
@pytest.mark.asyncio
async def test_admin_get_timetable_request_and_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
proto = MockProtocol()
client._protocol = proto
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_timetable(5))
await asyncio.sleep(0) # let the task send the request
# First use auto-subscribes to Gamescript updates, then sends the query.
assert len(proto.sent) == 2
assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency
assert decode_gamescript_payload(proto.sent[1]) == {
"command": "get_timetable", "vehicle_id": 5, "request_id": 1,
}
response = {"command": "get_timetable", "vehicle_id": 5, "request_id": 1,
"lateness": 0, "start_tick": 0, "current_order_time": 3,
"total_duration": -1, "orders": []}
await client.receive_ServerGamescript(None, data=response)
assert await task == response
assert client._gs_futures == {}
# Second call must not re-subscribe and must use a fresh request id.
task = asyncio.ensure_future(client.get_timetable(9))
await asyncio.sleep(0)
assert len(proto.sent) == 3
assert decode_gamescript_payload(proto.sent[2])["request_id"] == 2
await client.receive_ServerGamescript(None, data={"request_id": 2, "orders": []})
assert (await task)["orders"] == []
@pytest.mark.asyncio
async def test_admin_get_timetable_timeout():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
with pytest.raises(asyncio.TimeoutError):
await client.get_timetable(5, timeout=0.05)
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_timetable_error_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_timetable(65535))
await asyncio.sleep(0)
await client.receive_ServerGamescript(
None, data={"command": "get_timetable", "vehicle_id": 65535,
"request_id": 1, "error": "invalid_vehicle"})
with pytest.raises(ValueError, match="invalid_vehicle"):
await task
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_timetable_disconnect_fails_pending():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_timetable(5))
await asyncio.sleep(0)
client.disconnect(None)
with pytest.raises(ConnectionError):
await task
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_station_request_and_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
proto = MockProtocol()
client._protocol = proto
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_station(3))
await asyncio.sleep(0) # let the task send the request
# First use auto-subscribes to Gamescript updates, then sends the query.
assert len(proto.sent) == 2
assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency
assert decode_gamescript_payload(proto.sent[1]) == {
"command": "get_station", "station_id": 3, "request_id": 1,
}
response = {"command": "get_station", "station_id": 3, "request_id": 1,
"name": "Test Central", "location": 12345, "owner": 0,
"cargo": [{"cargo_id": 0, "waiting": 42, "planned": 17, "rating": 71}]}
await client.receive_ServerGamescript(None, data=response)
assert await task == response
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_station_error_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_station(65535))
await asyncio.sleep(0)
await client.receive_ServerGamescript(
None, data={"command": "get_station", "station_id": 65535,
"request_id": 1, "error": "invalid_station"})
with pytest.raises(ValueError, match="invalid_station"):
await task
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_station_timeout():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
with pytest.raises(asyncio.TimeoutError):
await client.get_station(3, timeout=0.05)
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_station_cargo_request_and_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
proto = MockProtocol()
client._protocol = proto
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_station_cargo(3, 0))
await asyncio.sleep(0) # let the task send the request
# First use auto-subscribes to Gamescript updates, then sends the query.
assert len(proto.sent) == 2
assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency
assert decode_gamescript_payload(proto.sent[1]) == {
"command": "get_station_cargo", "station_id": 3, "cargo_id": 0, "request_id": 1,
}
response = {"command": "get_station_cargo", "station_id": 3, "cargo_id": 0, "request_id": 1,
"waiting": 60, "planned": 40,
"waiting_by_from": [{"station": 5, "amount": 25}, {"station": 6, "amount": 35}],
"planned_by_from": [{"station": 5, "amount": 40}],
"waiting_by_via": [{"station": 7, "amount": 60}],
"planned_by_via": [{"station": 7, "amount": 40}]}
await client.receive_ServerGamescript(None, data=response)
assert await task == response
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_station_cargo_with_filters_encoding():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
proto = MockProtocol()
client._protocol = proto
client._transport = MockTransport()
client._gs_subscribed = True # skip the auto-subscribe so only the query is sent
task = asyncio.ensure_future(client.get_station_cargo(3, 0, from_station=6, via_station=7))
await asyncio.sleep(0)
assert len(proto.sent) == 1
assert decode_gamescript_payload(proto.sent[0]) == {
"command": "get_station_cargo", "station_id": 3, "cargo_id": 0,
"from_station": 6, "via_station": 7, "request_id": 1,
}
await client.receive_ServerGamescript(
None, data={"request_id": 1, "waiting": 12, "planned": 8})
assert (await task)["waiting"] == 12
@pytest.mark.asyncio
async def test_admin_get_station_cargo_error_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_station_cargo(3, 999))
await asyncio.sleep(0)
await client.receive_ServerGamescript(
None, data={"command": "get_station_cargo", "station_id": 3, "cargo_id": 999,
"request_id": 1, "error": "invalid_cargo"})
with pytest.raises(ValueError, match="invalid_cargo"):
await task
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_dispatch_request_and_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
proto = MockProtocol()
client._protocol = proto
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_dispatch(7))
await asyncio.sleep(0) # let the task send the request
# First use auto-subscribes to Gamescript updates, then sends the query.
assert len(proto.sent) == 2
assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency
assert decode_gamescript_payload(proto.sent[1]) == {
"command": "get_dispatch", "vehicle_id": 7, "request_id": 1,
}
response = {"command": "get_dispatch", "vehicle_id": 7, "request_id": 1,
"enabled": 1,
"schedules": [{"index": 0, "duration": 3000, "start_tick": 0, "delay": 0,
"reuse_slots": 0,
"slots": [{"offset": 500, "flags": 0}, {"offset": 1500, "flags": 0}]}]}
await client.receive_ServerGamescript(None, data=response)
assert await task == response
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_get_dispatch_error_response():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
task = asyncio.ensure_future(client.get_dispatch(65535))
await asyncio.sleep(0)
await client.receive_ServerGamescript(
None, data={"command": "get_dispatch", "vehicle_id": 65535,
"request_id": 1, "error": "invalid_vehicle"})
with pytest.raises(ValueError, match="invalid_vehicle"):
await task
assert client._gs_futures == {}
@pytest.mark.asyncio
async def test_admin_gamescript_passthrough_unmatched():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
client._protocol = MockProtocol()
client._transport = MockTransport()
gs_events = []
client.on_gamescript = lambda data: gs_events.append(data)
# No request_id, unknown request_id, and non-dict payloads all pass through.
await client.receive_ServerGamescript(None, data={"vehicles": []})
await client.receive_ServerGamescript(None, data={"request_id": 999, "orders": []})
await client.receive_ServerGamescript(None, data="plain string")
assert gs_events == [{"vehicles": []}, {"request_id": 999, "orders": []}, "plain string"]

View File

@@ -13,9 +13,19 @@ from openttd.protocol import (
OpenTTDAdminProtocol,
AdminUpdateType,
AdminUpdateFrequency,
PacketGameType
PacketGameType,
ModifyTimetableFlags
)
# These identify a vehicle/order that already exists in the local dev server's persisted
# save (company 0, unprotected, owns vehicle 7 with 2 orders) -- required for the timetable
# command tests below, since DoCommands are rejected unless issued by the owning company.
TIMETABLE_COMPANY_ID = 0
TIMETABLE_VEHICLE_ID = 7
TIMETABLE_ORDER_POSITION = 0
# A station TIMETABLE_VEHICLE_ID can legally serve, used for add_order/remove_order tests.
ORDER_STATION_ID = 6
# --- Pytest Fixtures ---
@@ -52,6 +62,22 @@ async def connected_client(server_config):
if hasattr(client, '_transport') and not client.shutdown_event.is_set():
await client.quit()
@pytest_asyncio.fixture
async def connected_owner_client(server_config):
"""Fixture to yield a client joined to TIMETABLE_COMPANY_ID (owns a real vehicle for command tests)."""
client_name = f"E2E_Owner_{random.randint(1000, 9999)}"
client = OpenTTDClient(
host=server_config["host"],
port=server_config["game_port"],
username=client_name
)
await client.connect(server_password=server_config["password"])
await client.join_company(company_id=TIMETABLE_COMPANY_ID, company_password="")
await asyncio.wait_for(client.joined.wait(), timeout=15.0)
yield client
if hasattr(client, '_transport') and not client.shutdown_event.is_set():
await client.quit()
# ==============================================================================
# --- End-to-End Tests (Covering all public functions with multiple inputs) ---
@@ -129,6 +155,222 @@ async def test_e2e_client_quit_and_disconnect_multiple_inputs(server_config):
# Input 2: quit already inactive client
await client2.quit()
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_change_timetable_wait_time(connected_owner_client):
# Public function: change_timetable()
# Input 1: set wait time
await connected_owner_client.change_timetable(TIMETABLE_VEHICLE_ID, TIMETABLE_ORDER_POSITION, ModifyTimetableFlags.WaitTime, 42)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
assert entry["orders"][TIMETABLE_ORDER_POSITION]["wait_time"] == 42
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_change_timetable_travel_time(connected_owner_client):
# Public function: change_timetable()
# Input 2: set travel time
await connected_owner_client.change_timetable(TIMETABLE_VEHICLE_ID, TIMETABLE_ORDER_POSITION, ModifyTimetableFlags.TravelTime, 99)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
assert entry["orders"][TIMETABLE_ORDER_POSITION]["travel_time"] == 99
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_autofill_timetable_enable(connected_owner_client):
# Public function: autofill_timetable()
# Input 1: enable autofill
await connected_owner_client.autofill_timetable(TIMETABLE_VEHICLE_ID, autofill=True, preserve_wait_time=False)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)["autofill"] is True
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_autofill_timetable_disable(connected_owner_client):
# Public function: autofill_timetable()
# Input 2: disable autofill, preserve wait time
await connected_owner_client.autofill_timetable(TIMETABLE_VEHICLE_ID, autofill=False, preserve_wait_time=True)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
assert entry["autofill"] is False
assert entry["autofill_preserve_wait_time"] is True
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_set_timetable_start_single_vehicle(connected_owner_client):
# Public function: set_timetable_start()
# Input 1: this vehicle only
await connected_owner_client.set_timetable_start(TIMETABLE_VEHICLE_ID, False, 500000)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
assert entry["timetable_start"] == 500000
assert entry["timetable_all"] is False
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_set_timetable_start_all_shared(connected_owner_client):
# Public function: set_timetable_start()
# Input 2: all vehicles sharing this order list
await connected_owner_client.set_timetable_start(TIMETABLE_VEHICLE_ID, True, 600000)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
assert entry["timetable_start"] == 600000
assert entry["timetable_all"] is True
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_set_vehicle_on_time_single_vehicle(connected_owner_client):
# Public function: set_vehicle_on_time()
# Input 1: reset lateness for this vehicle only
await connected_owner_client.set_vehicle_on_time(TIMETABLE_VEHICLE_ID, apply_to_group=False)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)["on_time_apply_to_group"] is False
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_set_vehicle_on_time_apply_to_group(connected_owner_client):
# Public function: set_vehicle_on_time()
# Input 2: reset lateness for every vehicle sharing these orders
await connected_owner_client.set_vehicle_on_time(TIMETABLE_VEHICLE_ID, apply_to_group=True)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)["on_time_apply_to_group"] is True
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_get_vehicle_timetable_after_change(connected_owner_client):
# Public function: get_vehicle_timetable()
# Input 1: a vehicle with observed state
await connected_owner_client.change_timetable(TIMETABLE_VEHICLE_ID, TIMETABLE_ORDER_POSITION, ModifyTimetableFlags.WaitTime, 15)
await asyncio.sleep(1.0)
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID) is not None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_get_vehicle_timetable_unknown_vehicle(connected_owner_client):
# Public function: get_vehicle_timetable()
# Input 2: a vehicle id with no observed state
assert connected_owner_client.get_vehicle_timetable(999999) is None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_add_and_remove_order(connected_owner_client, connected_admin):
# Public functions: add_order(), remove_order()
# Verified authoritatively via the admin get_timetable() order count. The test appends and
# inserts an order, then removes both, leaving the vehicle's order list as it started.
async def order_count():
data = await connected_admin.get_timetable(TIMETABLE_VEHICLE_ID, timeout=10.0)
return len(data["orders"])
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
before = await order_count()
# add_order input 1: append a goto-station order to the end of the list.
await connected_owner_client.add_order(TIMETABLE_VEHICLE_ID, ORDER_STATION_ID)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
assert await order_count() == before + 1
# add_order input 2: insert another before position 0.
await connected_owner_client.add_order(TIMETABLE_VEHICLE_ID, ORDER_STATION_ID, before_position=0)
await asyncio.sleep(1.0)
assert await order_count() == before + 2
# remove_order input 1: delete the one just inserted at the front.
await connected_owner_client.remove_order(TIMETABLE_VEHICLE_ID, 0)
await asyncio.sleep(1.0)
assert await order_count() == before + 1
# remove_order input 2: delete the appended order (now the last one) to restore the list.
await connected_owner_client.remove_order(TIMETABLE_VEHICLE_ID, before)
await asyncio.sleep(1.0)
assert not connected_owner_client.shutdown_event.is_set()
assert await order_count() == before
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_client_scheduled_dispatch_edit_and_view(connected_owner_client, connected_admin):
# Public functions: set_scheduled_dispatch(), add_dispatch_schedule(), remove_dispatch_schedule(),
# add_dispatch_slot(), remove_dispatch_slot(), clear_dispatch_schedule(), set_dispatch_duration(),
# set_dispatch_start_date(), and get_dispatch(). Edits go over the game port and are read back
# authoritatively via the admin get_dispatch(). The test leaves the vehicle with no schedules.
veh = TIMETABLE_VEHICLE_ID
owner = connected_owner_client
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
async def dispatch():
return await connected_admin.get_dispatch(veh, timeout=10.0)
start = await dispatch() # get_dispatch input 1: a valid vehicle
assert "schedules" in start and isinstance(start["schedules"], list)
if start["schedules"]:
pytest.skip("Test vehicle already has dispatch schedules; expected a clean vehicle.")
# add_dispatch_schedule: two schedules (indices 0 and 1) with different start ticks/durations.
await owner.add_dispatch_schedule(veh, 0, 3000)
await asyncio.sleep(0.5)
await owner.add_dispatch_schedule(veh, 1000, 2000)
await asyncio.sleep(0.5)
assert not owner.shutdown_event.is_set()
data = await dispatch()
assert len(data["schedules"]) == 2
assert data["schedules"][0]["duration"] == 3000
assert data["schedules"][1]["duration"] == 2000
# set_dispatch_duration / set_dispatch_start_date: two inputs each (schedule 0 and 1).
await owner.set_dispatch_duration(veh, 0, 4000)
await owner.set_dispatch_duration(veh, 1, 2500)
await owner.set_dispatch_start_date(veh, 0, 1_000_000)
await owner.set_dispatch_start_date(veh, 1, 2_000_000)
await asyncio.sleep(0.5)
# add_dispatch_slot: two departure slots in schedule 0.
await owner.add_dispatch_slot(veh, 0, 500)
await owner.add_dispatch_slot(veh, 0, 1500)
await asyncio.sleep(0.5)
data = await dispatch()
sched0 = data["schedules"][0]
assert sched0["duration"] == 4000
# The engine normalises the start tick relative to current game time (advancing it by whole
# durations to sit near "now"), so it won't equal the requested value verbatim; just confirm
# a start date was accepted and is reported as an integer.
assert isinstance(sched0["start_tick"], int)
assert {s["offset"] for s in sched0["slots"]} == {500, 1500}
# remove_dispatch_slot: two inputs (both slots of schedule 0).
await owner.remove_dispatch_slot(veh, 0, 1500)
await owner.remove_dispatch_slot(veh, 0, 500)
await asyncio.sleep(0.5)
assert (await dispatch())["schedules"][0]["slots"] == []
# set_scheduled_dispatch: enable then disable (two inputs), reading the flag back in between.
await owner.set_scheduled_dispatch(veh, True)
await asyncio.sleep(0.5)
assert (await dispatch())["enabled"] == 1
await owner.set_scheduled_dispatch(veh, False)
await asyncio.sleep(0.5)
assert (await dispatch())["enabled"] == 0
# clear_dispatch_schedule: two inputs (schedule 0 and 1).
await owner.clear_dispatch_schedule(veh, 0)
await owner.clear_dispatch_schedule(veh, 1)
await asyncio.sleep(0.5)
# remove_dispatch_schedule: remove both (higher index first) to restore the vehicle.
await owner.remove_dispatch_schedule(veh, 1)
await asyncio.sleep(0.5)
await owner.remove_dispatch_schedule(veh, 0)
await asyncio.sleep(0.5)
assert not owner.shutdown_event.is_set()
assert (await dispatch())["schedules"] == [] # get_dispatch input 1 (restored state)
# --- Admin Client Public Functions ---
@@ -301,6 +543,176 @@ async def test_e2e_admin_list_vehicles_specific_company(connected_admin):
assert "vehicles" in responses[-1]
assert isinstance(responses[-1]["vehicles"], list)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_get_timetable_valid_vehicle(connected_admin):
# Public function: get_timetable()
# Input 1: a real vehicle id discovered via list_vehicles
responses = []
connected_admin.on_gamescript = lambda data: responses.append(data)
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
await connected_admin.list_vehicles()
await asyncio.sleep(0.5)
assert len(responses) >= 1 and "vehicles" in responses[-1]
vehicles = responses[-1]["vehicles"]
if not vehicles:
pytest.skip("No vehicles on the test server to query a timetable for.")
vid = vehicles[0]["id"]
data = await connected_admin.get_timetable(vid, timeout=10.0)
assert data["vehicle_id"] == vid
for key in ("lateness", "start_tick", "current_order_time", "total_duration", "orders"):
assert key in data
assert isinstance(data["orders"], list)
for order in data["orders"]:
for key in ("position", "wait_time", "travel_time", "wait_timetabled",
"travel_timetabled", "wait_fixed", "travel_fixed", "leave_type", "max_speed"):
assert key in order
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_get_timetable_invalid_vehicle(connected_admin):
# Public function: get_timetable()
# Input 2: an id no vehicle can have -> GameScript reports invalid_vehicle
with pytest.raises(ValueError, match="invalid_vehicle"):
await connected_admin.get_timetable(65535, timeout=10.0)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_get_dispatch_invalid_vehicle(connected_admin):
# Public function: get_dispatch()
# Input 2: an id no vehicle can have -> GameScript reports invalid_vehicle
with pytest.raises(ValueError, match="invalid_vehicle"):
await connected_admin.get_dispatch(65535, timeout=10.0)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_list_stations_all_companies(connected_admin):
# Public function: list_stations()
# Input 1: all companies (no company_id)
responses = []
connected_admin.on_gamescript = lambda data: responses.append(data)
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
await connected_admin.list_stations()
await asyncio.sleep(0.5)
assert not connected_admin.shutdown_event.is_set()
assert len(responses) >= 1
assert "stations" in responses[-1]
assert isinstance(responses[-1]["stations"], list)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_list_stations_specific_company(connected_admin):
# Public function: list_stations()
# Input 2: specific company_id
responses = []
connected_admin.on_gamescript = lambda data: responses.append(data)
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
await connected_admin.list_stations(company_id=0)
await asyncio.sleep(0.5)
assert not connected_admin.shutdown_event.is_set()
assert len(responses) >= 1
assert "stations" in responses[-1]
assert isinstance(responses[-1]["stations"], list)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_get_station_valid_station(connected_admin):
# Public function: get_station()
# Input 1: a real station id discovered via list_stations
responses = []
connected_admin.on_gamescript = lambda data: responses.append(data)
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
await connected_admin.list_stations()
await asyncio.sleep(0.5)
assert len(responses) >= 1 and "stations" in responses[-1]
stations = responses[-1]["stations"]
if not stations:
pytest.skip("No stations on the test server to query.")
sid = stations[0]["id"]
data = await connected_admin.get_station(sid, timeout=10.0)
assert data["station_id"] == sid
assert "cargo" in data
assert isinstance(data["cargo"], list)
for cargo in data["cargo"]:
for key in ("cargo_id", "waiting", "planned", "rating"):
assert key in cargo
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_get_station_invalid_station(connected_admin):
# Public function: get_station()
# Input 2: an id no station can have -> GameScript reports invalid_station
with pytest.raises(ValueError, match="invalid_station"):
await connected_admin.get_station(65535, timeout=10.0)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_get_station_cargo_breakdown(connected_admin):
# Public function: get_station_cargo()
# Input 1: a real station + a cargo it has handled, discovered via list_stations/get_station
responses = []
connected_admin.on_gamescript = lambda data: responses.append(data)
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
await connected_admin.list_stations()
await asyncio.sleep(0.5)
assert responses and "stations" in responses[-1]
stations = responses[-1]["stations"]
if not stations:
pytest.skip("No stations on the test server to query.")
# Find a station/cargo pair that actually has cargo data.
target = None
for st in stations:
detail = await connected_admin.get_station(st["id"], timeout=10.0)
if detail["cargo"]:
target = (st["id"], detail["cargo"][0]["cargo_id"])
break
if target is None:
pytest.skip("No station with handled cargo to break down.")
sid, cid = target
data = await connected_admin.get_station_cargo(sid, cid, timeout=10.0)
assert data["station_id"] == sid and data["cargo_id"] == cid
for key in ("waiting", "planned",
"waiting_by_from", "planned_by_from", "waiting_by_via", "planned_by_via"):
assert key in data
for key in ("waiting_by_from", "planned_by_from", "waiting_by_via", "planned_by_via"):
assert isinstance(data[key], list)
for entry in data[key]:
assert "station" in entry and "amount" in entry
# Input 2: the same query narrowed by a next-hop (via) filter is accepted and echoes it back.
filtered = await connected_admin.get_station_cargo(sid, cid, via_station=sid, timeout=10.0)
assert filtered["via_station"] == sid
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_get_station_cargo_invalid_cargo(connected_admin):
# Public function: get_station_cargo()
# A cargo id no cargo can have -> GameScript reports invalid_cargo. Needs a valid station.
responses = []
connected_admin.on_gamescript = lambda data: responses.append(data)
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
await connected_admin.list_stations()
await asyncio.sleep(0.5)
assert responses and "stations" in responses[-1]
stations = responses[-1]["stations"]
if not stations:
pytest.skip("No stations on the test server to query.")
with pytest.raises(ValueError, match="invalid_cargo"):
await connected_admin.get_station_cargo(stations[0]["id"], 250, timeout=10.0)
# --- Protocol Public Functions ---

View File

@@ -118,7 +118,6 @@ async def test_fallback_handlers():
await client.receive_ServerConfigurationUpdate(None)
await client.receive_ServerClientInfo(None)
await client.receive_ServerExternalChat(None)
await client.receive_ServerCommand(None)
await client.receive_ClientAck(None)
await client.receive_ClientIdentify(None)
await client.receive_ServerCompanyUpdate(None)
@@ -224,7 +223,6 @@ async def test_unit_client_noop_callbacks(server_config):
await client.receive_ServerMapData(None)
await client.receive_ServerConfigurationUpdate(None)
await client.receive_ServerExternalChat(None)
await client.receive_ServerCommand(None)
await client.receive_ServerFull(None)
await client.receive_ServerBanned(None)
await client.receive_ClientAck(None)

View File

@@ -38,7 +38,6 @@ def test_protocol_static_parsers():
assert res["token"] == 7
assert OpenTTDProtocol.receive_ServerExternalChat(None, b"") == {}
assert OpenTTDProtocol.receive_ServerCommand(None, b"") == {}
assert OpenTTDProtocol.receive_ServerFull(None, b"") == {}
assert OpenTTDProtocol.receive_ServerBanned(None, b"") == {}
assert OpenTTDProtocol.receive_ClientIdentify(None, b"") == {}

426
tests/test_timetable.py Normal file
View File

@@ -0,0 +1,426 @@
import pytest
from openttd import OpenTTDClient
from openttd.protocol import (
OpenTTDProtocol, GameCommand, ModifyTimetableFlags, ModifyTimetableCtrlFlag,
OrderType, OrderNonStopFlags, OrderStopLocation, INVALID_VEH_ORDER_ID,
write_varuint, read_varuint, write_varuint_signed, read_varuint_signed
)
from openttd_protocol.wire.read import read_uint8, read_uint16
class MockTransport:
def is_closing(self):
return False
def write(self, data):
return len(data)
class MockProtocol:
def __init__(self):
self.sent = []
async def send_packet(self, data):
self.sent.append(data)
return len(data)
def decode_sent_command(packet):
"""Decode a ClientCommand packet by reusing the ServerCommand parser, padding on the
frame/my_cmd trailer that only ServerCommand carries on the wire (ClientCommand doesn't)."""
padded = bytes(packet)[3:] + b"\x00\x00\x00\x00\x00"
return OpenTTDProtocol.receive_ServerCommand(None, memoryview(padded))
def new_client():
client = OpenTTDClient("127.0.0.1")
client._protocol = MockProtocol()
client._transport = MockTransport()
client._target_company = 0
return client
# --- Varuint codec ---
@pytest.mark.parametrize("value", [
0, 1, 127,
128, 16383,
16384, 2097151,
2097152, 268435455,
268435456, 34359738367,
34359738368, 4398046511103,
4398046511104, 562949953421311,
562949953421312, 72057594037927935,
72057594037927936, 18446744073709551615,
])
def test_varuint_roundtrip_boundaries(value):
buf = bytearray()
write_varuint(buf, value)
decoded, rest = read_varuint(memoryview(bytes(buf)))
assert decoded == value
assert bytes(rest) == b""
def test_varuint_rejects_negative():
with pytest.raises(ValueError):
write_varuint(bytearray(), -1)
@pytest.mark.parametrize("value", [0, 1, -1, 2, -2, 1000000, -1000000, 9223372036854775807, -9223372036854775808])
def test_varuint_signed_roundtrip(value):
buf = bytearray()
write_varuint_signed(buf, value)
decoded, rest = read_varuint_signed(memoryview(bytes(buf)))
assert decoded == value
assert bytes(rest) == b""
# --- OpenTTDProtocol.receive_ServerCommand ---
def build_server_command_bytes(company, cmd, payload, callback=0, callback_param=0, frame=42, my_cmd=True):
import struct
body = bytearray()
body += struct.pack("<B", company)
body += struct.pack("<H", cmd)
body += struct.pack("<H", 0)
body += struct.pack("<I", 0)
body += struct.pack("<H", len(payload))
body += payload
body += struct.pack("<B", callback)
if callback != 0:
body += struct.pack("<I", callback_param)
body += struct.pack("<I", frame)
body += struct.pack("<B", 1 if my_cmd else 0)
return bytes(body)
def test_protocol_receive_server_command_no_callback():
payload = bytearray()
write_varuint(payload, 7)
data = build_server_command_bytes(1, GameCommand.ChangeTimetable, payload, callback=0, frame=42, my_cmd=True)
res = OpenTTDProtocol.receive_ServerCommand(None, memoryview(data))
assert res["company"] == 1
assert res["cmd"] == GameCommand.ChangeTimetable
assert res["callback"] == 0
assert res["callback_param"] == 0
assert res["frame"] == 42
assert res["my_cmd"] is True
assert bytes(res["payload"]) == bytes(payload)
def test_protocol_receive_server_command_with_callback():
payload = bytearray()
write_varuint(payload, 9)
data = build_server_command_bytes(2, GameCommand.SetVehicleOnTime, payload, callback=5, callback_param=999, frame=100, my_cmd=False)
res = OpenTTDProtocol.receive_ServerCommand(None, memoryview(data))
assert res["callback"] == 5
assert res["callback_param"] == 999
assert res["my_cmd"] is False
# --- OpenTTDClient outgoing command methods ---
@pytest.mark.asyncio
async def test_client_send_command_with_callback_includes_callback_param():
client = new_client()
await client._send_command(GameCommand.ChangeTimetable, bytearray(), callback=5)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["callback"] == 5
assert parsed["callback_param"] == 0
@pytest.mark.asyncio
async def test_client_change_timetable_sends_expected_payload():
client = new_client()
await client.change_timetable(7, 3, ModifyTimetableFlags.WaitTime, 120)
assert len(client._protocol.sent) == 1
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.ChangeTimetable
assert parsed["company"] == 0
vehicle_id, rest = read_varuint(parsed["payload"])
order_position, rest = read_uint16(rest)
flag, rest = read_uint8(rest)
value, rest = read_varuint(rest)
ctrl_flags, _ = read_uint8(rest)
assert (vehicle_id, order_position, flag, value, ctrl_flags) == (7, 3, ModifyTimetableFlags.WaitTime, 120, 0)
@pytest.mark.asyncio
async def test_client_change_timetable_clear_field_sets_ctrl_flag():
client = new_client()
await client.change_timetable(7, 3, ModifyTimetableFlags.TravelTime, 0, clear_field=True)
parsed = decode_sent_command(client._protocol.sent[0])
vehicle_id, rest = read_varuint(parsed["payload"])
order_position, rest = read_uint16(rest)
flag, rest = read_uint8(rest)
value, rest = read_varuint(rest)
ctrl_flags, _ = read_uint8(rest)
assert flag == ModifyTimetableFlags.TravelTime
assert ctrl_flags == ModifyTimetableCtrlFlag.ClearField
@pytest.mark.asyncio
async def test_client_autofill_timetable_sends_expected_payload():
client = new_client()
await client.autofill_timetable(7, autofill=True, preserve_wait_time=False)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.AutofillTimetable
vehicle_id, rest = read_varuint(parsed["payload"])
autofill, rest = read_uint8(rest)
preserve_wait_time, _ = read_uint8(rest)
assert (vehicle_id, autofill, preserve_wait_time) == (7, 1, 0)
@pytest.mark.asyncio
async def test_client_set_timetable_start_sends_expected_payload():
client = new_client()
await client.set_timetable_start(7, True, -12345)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SetTimetableStart
vehicle_id, rest = read_varuint(parsed["payload"])
timetable_all, rest = read_uint8(rest)
start_date, _ = read_varuint_signed(rest)
assert (vehicle_id, timetable_all, start_date) == (7, 1, -12345)
@pytest.mark.asyncio
async def test_client_set_vehicle_on_time_sends_expected_payload():
client = new_client()
await client.set_vehicle_on_time(7, apply_to_group=True)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SetVehicleOnTime
vehicle_id, rest = read_varuint(parsed["payload"])
apply_to_group, _ = read_uint8(rest)
assert (vehicle_id, apply_to_group) == (7, 1)
# --- OpenTTDClient order add/remove commands ---
def _decode_insert_order_payload(payload):
vehicle_id, rest = read_varuint(payload)
sel_ord, rest = read_uint16(rest)
order_type, rest = read_uint8(rest)
order_flags, rest = read_uint16(rest)
station, _ = read_uint16(rest)
return vehicle_id, sel_ord, order_type, order_flags, station
@pytest.mark.asyncio
async def test_client_add_order_appends_by_default():
client = new_client()
await client.add_order(7, 6)
assert len(client._protocol.sent) == 1
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.InsertOrder
assert parsed["company"] == 0
vehicle_id, sel_ord, order_type, order_flags, station = _decode_insert_order_payload(parsed["payload"])
assert vehicle_id == 7
assert sel_ord == INVALID_VEH_ORDER_ID # append to the end
# OT_GOTO_STATION in bits 0-3, far-end stop location in bits 4-5, stop-everywhere non-stop in 6-7
assert order_type == (OrderType.GotoStation | (OrderStopLocation.PlatformFarEnd << 4))
assert order_flags == 0
assert station == 6
@pytest.mark.asyncio
async def test_client_add_order_insert_position_and_nonstop():
client = new_client()
await client.add_order(7, 6, before_position=1, non_stop=OrderNonStopFlags.NoStopAtIntermediate)
parsed = decode_sent_command(client._protocol.sent[0])
vehicle_id, sel_ord, order_type, order_flags, station = _decode_insert_order_payload(parsed["payload"])
assert sel_ord == 1 # insert before order position 1
assert order_type == (OrderType.GotoStation
| (OrderStopLocation.PlatformFarEnd << 4)
| (OrderNonStopFlags.NoStopAtIntermediate << 6))
@pytest.mark.asyncio
async def test_client_remove_order_sends_expected_payload():
client = new_client()
await client.remove_order(7, 2)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.DeleteOrder
vehicle_id, rest = read_varuint(parsed["payload"])
order_position, _ = read_uint16(rest)
assert (vehicle_id, order_position) == (7, 2)
# --- OpenTTDClient scheduled dispatch edit commands ---
@pytest.mark.asyncio
async def test_client_set_scheduled_dispatch_payload():
client = new_client()
await client.set_scheduled_dispatch(7, True)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SchDispatch
vehicle_id, rest = read_varuint(parsed["payload"])
enabled, _ = read_uint8(rest)
assert (vehicle_id, enabled) == (7, 1)
@pytest.mark.asyncio
async def test_client_add_dispatch_schedule_payload():
client = new_client()
await client.add_dispatch_schedule(7, -1234, 3000)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SchDispatchAddNewSchedule
vehicle_id, rest = read_varuint(parsed["payload"])
start_tick, rest = read_varuint_signed(rest)
duration, _ = read_varuint(rest)
assert (vehicle_id, start_tick, duration) == (7, -1234, 3000)
@pytest.mark.asyncio
async def test_client_remove_dispatch_schedule_payload():
client = new_client()
await client.remove_dispatch_schedule(7, 2)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SchDispatchRemoveSchedule
vehicle_id, rest = read_varuint(parsed["payload"])
schedule_index, _ = read_varuint(rest)
assert (vehicle_id, schedule_index) == (7, 2)
@pytest.mark.asyncio
async def test_client_add_dispatch_slot_payload_defaults_and_extras():
client = new_client()
# Defaults: single slot, no interval/extra/flags/route.
await client.add_dispatch_slot(7, 1, 500)
# Bulk: three extra slots spaced 250 ticks apart, with flags and route id.
await client.add_dispatch_slot(7, 1, 500, interval=250, extra_slots=3, slot_flags=5, route_id=2)
def decode(payload):
vehicle_id, rest = read_varuint(payload)
schedule_index, rest = read_varuint(rest)
offset, rest = read_varuint(rest)
interval, rest = read_varuint(rest)
extra_slots, rest = read_varuint(rest)
slot_flags, rest = read_uint16(rest)
route_id, _ = read_uint8(rest)
return (vehicle_id, schedule_index, offset, interval, extra_slots, slot_flags, route_id)
p0 = decode_sent_command(client._protocol.sent[0])
p1 = decode_sent_command(client._protocol.sent[1])
assert p0["cmd"] == GameCommand.SchDispatchAdd
assert decode(p0["payload"]) == (7, 1, 500, 0, 0, 0, 0)
assert decode(p1["payload"]) == (7, 1, 500, 250, 3, 5, 2)
@pytest.mark.asyncio
async def test_client_remove_dispatch_slot_payload():
client = new_client()
await client.remove_dispatch_slot(7, 1, 500)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SchDispatchRemove
vehicle_id, rest = read_varuint(parsed["payload"])
schedule_index, rest = read_varuint(rest)
offset, _ = read_varuint(rest)
assert (vehicle_id, schedule_index, offset) == (7, 1, 500)
@pytest.mark.asyncio
async def test_client_clear_dispatch_schedule_payload():
client = new_client()
await client.clear_dispatch_schedule(7, 1)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SchDispatchClear
vehicle_id, rest = read_varuint(parsed["payload"])
schedule_index, _ = read_varuint(rest)
assert (vehicle_id, schedule_index) == (7, 1)
@pytest.mark.asyncio
async def test_client_set_dispatch_duration_payload():
client = new_client()
await client.set_dispatch_duration(7, 1, 4000)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SchDispatchSetDuration
vehicle_id, rest = read_varuint(parsed["payload"])
schedule_index, rest = read_varuint(rest)
duration, _ = read_varuint(rest)
assert (vehicle_id, schedule_index, duration) == (7, 1, 4000)
@pytest.mark.asyncio
async def test_client_set_dispatch_start_date_payload():
client = new_client()
await client.set_dispatch_start_date(7, 1, 1_000_000)
parsed = decode_sent_command(client._protocol.sent[0])
assert parsed["cmd"] == GameCommand.SchDispatchSetStartDate
vehicle_id, rest = read_varuint(parsed["payload"])
schedule_index, rest = read_varuint(rest)
start_tick, _ = read_varuint_signed(rest)
assert (vehicle_id, schedule_index, start_tick) == (7, 1, 1_000_000)
# --- OpenTTDClient.receive_ServerCommand dispatch ---
async def feed_command(client, cmd, payload):
parsed = OpenTTDProtocol.receive_ServerCommand(None, memoryview(build_server_command_bytes(0, cmd, payload)))
await client.receive_ServerCommand(None, **parsed)
@pytest.mark.asyncio
async def test_receive_change_timetable_updates_order_state():
client = new_client()
payload = bytearray()
write_varuint(payload, 7)
payload += (3).to_bytes(2, "little")
payload.append(ModifyTimetableFlags.WaitTime)
write_varuint(payload, 120)
payload.append(0)
await feed_command(client, GameCommand.ChangeTimetable, payload)
assert client.get_vehicle_timetable(7) == {"orders": {3: {"wait_time": 120}}}
@pytest.mark.asyncio
async def test_receive_change_timetable_clear_field_sets_none():
client = new_client()
payload = bytearray()
write_varuint(payload, 7)
payload += (3).to_bytes(2, "little")
payload.append(ModifyTimetableFlags.TravelTime)
write_varuint(payload, 0)
payload.append(ModifyTimetableCtrlFlag.ClearField)
await feed_command(client, GameCommand.ChangeTimetable, payload)
assert client.get_vehicle_timetable(7)["orders"][3]["travel_time"] is None
@pytest.mark.asyncio
async def test_receive_change_timetable_wait_fixed_stores_bool():
client = new_client()
payload = bytearray()
write_varuint(payload, 7)
payload += (0).to_bytes(2, "little")
payload.append(ModifyTimetableFlags.SetWaitFixed)
write_varuint(payload, 1)
payload.append(0)
await feed_command(client, GameCommand.ChangeTimetable, payload)
assert client.get_vehicle_timetable(7)["orders"][0]["wait_time_fixed"] is True
@pytest.mark.asyncio
async def test_receive_autofill_timetable_updates_state():
client = new_client()
payload = bytearray()
write_varuint(payload, 7)
payload.append(1)
payload.append(0)
await feed_command(client, GameCommand.AutofillTimetable, payload)
entry = client.get_vehicle_timetable(7)
assert entry["autofill"] is True
assert entry["autofill_preserve_wait_time"] is False
@pytest.mark.asyncio
async def test_receive_set_timetable_start_updates_state():
client = new_client()
payload = bytearray()
write_varuint(payload, 7)
payload.append(1)
write_varuint_signed(payload, 555)
await feed_command(client, GameCommand.SetTimetableStart, payload)
entry = client.get_vehicle_timetable(7)
assert entry["timetable_all"] is True
assert entry["timetable_start"] == 555
@pytest.mark.asyncio
async def test_receive_set_vehicle_on_time_updates_state():
client = new_client()
payload = bytearray()
write_varuint(payload, 7)
payload.append(1)
await feed_command(client, GameCommand.SetVehicleOnTime, payload)
assert client.get_vehicle_timetable(7)["on_time_apply_to_group"] is True
@pytest.mark.asyncio
async def test_receive_unknown_command_is_ignored():
client = new_client()
await feed_command(client, 999, bytearray())
assert client.vehicle_timetables == {}
# --- get_vehicle_timetable ---
@pytest.mark.asyncio
async def test_get_vehicle_timetable_known_and_unknown():
client = new_client()
assert client.get_vehicle_timetable(7) is None
payload = bytearray()
write_varuint(payload, 7)
payload.append(1)
await feed_command(client, GameCommand.SetVehicleOnTime, payload)
assert client.get_vehicle_timetable(7) is not None
assert client.get_vehicle_timetable(42) is None