README.md

@@ -10,6 +10,7 @@ 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.
 
 ## 🛠 Setup
 
@@ -73,4 +74,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)

docs/PROTOCOL.md

@@ -31,6 +31,28 @@ 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.
 
+## Vehicle Timetables (Game Port DoCommands)
+Unlike vehicle listing, timetables have no GameScript API surface at all (no getters or setters exist in `GSOrder`/`GSVehicle`). 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.
+
+### Timetable command IDs and payload tuples
+| Method | `cmd` | payload |
+|---|---|---|
+| `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)` |
+
+`VehicleID` and other 4/8-byte fields use OpenTTD's custom varuint scheme (`write_varuint`/`read_varuint` in `protocol.py`) — a UTF-8-like prefix encoding, not LEB128; signed fields (`StateTicks`) use zigzag on top of it (`write_varuint_signed`/`read_varuint_signed`).
+
+### Ownership requirement
+A command is rejected (and the client kicked) 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 timetable method.
+
+### Reading timetables — no query command exists
+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.
+
 ## Stream Encryption (AEAD)
 Once `ServerEnableEncryption` is received, all subsequent packets use **XChaCha20-Poly1305** (Authenticated Encryption with Associated Data).
 

docs/TIMETABLES.md

@@ -0,0 +1,275 @@
+# Vehicle Timetables: Usage Guide
+
+This guide covers `OpenTTDClient`'s timetable API: `change_timetable()`, `autofill_timetable()`,
+`set_timetable_start()`, `set_vehicle_on_time()`, and `get_vehicle_timetable()`. 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.
+
+## The two things you must know before using this API
+
+1. **You must join the vehicle's own company.** These are real game commands (`DoCommand`s), 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. **There is no "get timetable" query.** OpenTTD's protocol has no command that returns a
+   vehicle's current timetable. `get_vehicle_timetable()` works by passively watching the
+   `ServerCommand` broadcasts the server sends to every joined client whenever *anyone* changes a
+   timetable. This means:
+   - 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())
+```
+
+## See also
+- [PROTOCOL.md — Vehicle Timetables](PROTOCOL.md#vehicle-timetables-game-port-docommands) for the underlying wire format.
+- [ARCHITECTURE.md](ARCHITECTURE.md) for how `OpenTTDClient` fits into the rest of the library.

lib/openttd/client.py

@@ -5,7 +5,12 @@ 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,
+    write_varuint, read_varuint, write_varuint_signed, read_varuint_signed
+)
 from .decorators import exclude_call_check
 
 class OpenTTDClient:
@@ -22,6 +27,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 +68,68 @@ 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)
+
     def disconnect(self, source):
         """Library callback for when connection is lost."""
         self.log.info("Disconnected.")
@@ -199,7 +267,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

lib/openttd/protocol.py

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

main.py

@@ -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,95 @@ 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
+
+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)")
+
+    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 +131,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)

tests/test_e2e.py

@@ -13,9 +13,17 @@ 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
+
 
 # --- Pytest Fixtures ---
 
@@ -52,6 +60,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 +153,110 @@ 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
+
 
 # --- Admin Client Public Functions ---
 

tests/test_logic.py

@@ -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)

tests/test_protocol.py

@@ -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"") == {}

tests/test_timetable.py

@@ -0,0 +1,276 @@
+import pytest
+from openttd import OpenTTDClient
+from openttd.protocol import (
+    OpenTTDProtocol, GameCommand, ModifyTimetableFlags, ModifyTimetableCtrlFlag,
+    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.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