From d87c779d94830c04edf09ae4d4154058fae114c6 Mon Sep 17 00:00:00 2001 From: kovagoadi Date: Mon, 31 Aug 2026 18:14:40 +0200 Subject: [PATCH] Add game event support to the admin client Everything on the admin GameScript channel so far has been request/reply. This adds the other direction: subscribe_events() opens a push stream so a bot can react to the game instead of polling it, consumed either by awaiting wait_for_event() or via an on_event callback. Both see every event; an event goes to at most one waiter, and unclaimed ones sit in a bounded buffer. Sixteen kinds, from two sources. The engine raises no GameScript event for a vehicle reaching a stop or cargo arriving, so vehicle_arrive, vehicle_depart and cargo_waiting are synthesised by the bridge sampling state every `interval` ticks and diffing against the previous sample -- which means a stop shorter than the interval is never reported, and the first sample only establishes a baseline. The rest (crashes, industries, towns, companies, subsidies) are engine events forwarded verbatim. vehicle_lost, vehicle_waiting_in_depot and vehicle_unprofitable are deliberately absent: the engine raises those only for AI companies, so a GameScript can never observe them. The server-side half lives in the AdminBridge GameScript, which is not in this repo -- docker/config is gitignored -- so it has to be updated separately for any of this to work. Also repoints the scheduled-dispatch E2E test at a dedicated vehicle (DISPATCH_VEHICLE_ID). It had been silently skipping because vehicle 7 carries a hand-built annual dispatch schedule, which left eight dispatch methods unverified end to end while check_public_calls.py reported them green off static analysis of the call sites. Co-Authored-By: Claude --- README.md | 2 + docs/ARCHITECTURE.md | 5 + docs/EVENTS.md | 232 +++++++++++++++++++++++++++++++++++++ docs/PROTOCOL.md | 10 ++ docs/TESTING.md | 1 + lib/openttd/client.py | 151 +++++++++++++++++++++++- lib/openttd/protocol.py | 36 +++++- main_admin.py | 23 +++- tests/test_e2e.py | 99 +++++++++++++++- tests/test_events.py | 251 ++++++++++++++++++++++++++++++++++++++++ 10 files changed, 800 insertions(+), 10 deletions(-) create mode 100644 docs/EVENTS.md create mode 100644 tests/test_events.py diff --git a/README.md b/README.md index 8772783..4ad00d8 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ A high-performance, Object-Oriented Python client for OpenTTD servers, specifica - **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. +- **Game Events:** react to the game instead of polling it — `subscribe_events()` streams events over the AdminBridge GS as they happen: a vehicle reaching or leaving a stop (`vehicle_arrive`/`vehicle_depart`, with dwell time and cargo aboard), a station's waiting cargo changing (`cargo_waiting`), plus crashes, industries opening/closing, towns, companies and subsidies. Consume them with `await wait_for_event()` or an `on_event` callback; filter by kind, company, vehicle, station or cargo. ## 🛠 Setup @@ -81,4 +82,5 @@ For detailed instructions on E2E testing and coverage reports, see the [Testing - [Architecture & Design](docs/ARCHITECTURE.md) - [Protocol Internals (PAKE/Encryption)](docs/PROTOCOL.md) - [Vehicle Timetables Usage Guide](docs/TIMETABLES.md) +- [Game Events Usage Guide](docs/EVENTS.md) - [Contributor Guide](docs/CONTRIBUTING.md) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index eca96a6..48e6210 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -16,6 +16,11 @@ The primary API for developers. - **Event-Driven:** Uses `asyncio.Event` (like `self.joined`) to signal state changes to the calling code. - **Callback System:** Provides hooks like `on_chat` to allow users to react to game events without modifying the core library. +### 3. `OpenTTDAdminClient` (Admin Network) +Talks to the Admin port (TCP 3977) and, through the AdminBridge GameScript's JSON channel, to the running game itself. +- **Correlated Queries:** `get_timetable()`, `get_station()`, `get_dispatch()` and friends all funnel through `_gs_query()`, which tags each request with a monotonic `request_id`, parks a future, and lets `receive_ServerGamescript` resolve it when the matching reply arrives. +- **Push Events:** `subscribe_events()` opens the one stream that flows the other way. Event batches carry no `request_id`, so `receive_ServerGamescript` routes them to the event consumers instead: each event goes to the longest-waiting matching `wait_for_event()` caller, or into a bounded buffer if nobody is waiting, and to the `on_event` observer either way. See [EVENTS.md](EVENTS.md). + ## Handshake Flow 1. **Connection:** TCP connection established to Port 3979. diff --git a/docs/EVENTS.md b/docs/EVENTS.md new file mode 100644 index 0000000..b5f9a5b --- /dev/null +++ b/docs/EVENTS.md @@ -0,0 +1,232 @@ +# Game Events: Usage Guide + +Everything else in this library is a **request**: you ask the server something and it answers. +Events are the other direction — the server tells you when something happens, so a bot can react +to the game instead of polling it in a loop. This guide covers `OpenTTDAdminClient`'s +`subscribe_events()`, `unsubscribe_events()`, `wait_for_event()` and the `on_event` callback. +For the JSON that goes over the wire, see [PROTOCOL.md](PROTOCOL.md#game-events). + +## Quick start + +```python +import asyncio +from openttd import OpenTTDAdminClient +from openttd.protocol import GameEventType + +async def main(): + admin = OpenTTDAdminClient("127.0.0.1", admin_name="EventWatcher") + await admin.connect(admin_password="asd", secure=True) + await admin.joined.wait() + + await admin.subscribe_events( + events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart], + ) + + for _ in range(10): + event = await admin.wait_for_event(timeout=60.0) + print(f"vehicle {event['vehicle_id']} {event['event']} " + f"at station {event['station_id']}") + + await admin.unsubscribe_events() + await admin.quit() + +asyncio.run(main()) +``` + +## Two ways to consume events + +Both see **every** event, so you can use either or mix them. + +**Pull — `await admin.wait_for_event(kind=None, timeout=5.0)`.** Returns the next matching event. +`kind` is a `GameEventType` (or plain string), an iterable of them, or `None` for "anything". +Events that arrive while you are not waiting are buffered, so nothing is lost between two calls, +and each event goes to at most one waiter. Raises `asyncio.TimeoutError` if nothing matching +arrives in time, `ConnectionError` if the admin connection drops mid-wait. + +```python +# The next arrival or departure, whichever comes first. +event = await admin.wait_for_event( + {GameEventType.VehicleArrive, GameEventType.VehicleDepart}, timeout=30.0) +``` + +**Push — `admin.on_event = handler`.** A plain callback (like `on_chat` / `on_console`), invoked +with each event dict as it arrives. Use it for a long-running loop that reacts to everything: + +```python +def handle(event): + if event["event"] == "cargo_waiting" and event["delta"] > 0: + print(f"{event['delta']} units of cargo {event['cargo_id']} " + f"arrived at station {event['station_id']}") + +admin.on_event = handle +await admin.subscribe_events(events=[GameEventType.CargoWaiting]) +await asyncio.sleep(600) # events arrive on the connection's own task +``` + +The buffer behind `wait_for_event()` holds the last 256 unclaimed events by default +(`OpenTTDAdminClient(..., event_buffer_size=N)`); past that the oldest are dropped, so a +subscription nobody drains cannot grow without bound. + +## Event kinds + +Import them from `openttd.protocol` as `GameEventType`, or just use the string. They come in two +flavours, which behave differently in ways worth knowing. + +### Polled events — sampled, not instantaneous + +The engine raises **no** event when a vehicle reaches a stop or when cargo shows up, so the +AdminBridge GameScript synthesises these by sampling game state every `interval` ticks and +reporting what changed since the previous sample. + +| Kind | Fires when | Fields | +|---|---|---| +| `vehicle_arrive` | a vehicle starts loading/unloading at a station | `vehicle_id`, `station_id`, `owner`, `vehicle_type`, `order_position`, `cargo` | +| `vehicle_depart` | a vehicle stops loading/unloading at a station | the same, plus `dwell` (ticks it stayed) | +| `cargo_waiting` | a station's waiting amount of a cargo changes | `station_id`, `cargo_id`, `waiting` (new total), `delta` (signed change) | + +`cargo` is a list of `{"cargo_id": n, "load": units}` for the cargo actually aboard (omit it with +`include_cargo=False`). `vehicle_type` is `0` rail, `1` road, `2` water, `3` air. +`order_position` is the index of the vehicle's current order, or `-1` if it has none. +`delta` is negative when cargo was picked up rather than delivered. + +Three consequences of sampling that you should design around: + +- **Short stops can be missed entirely.** If a vehicle arrives and leaves between two samples, + neither event exists. Lower `interval` to narrow the window; you cannot close it. +- **The first sample only establishes a baseline.** A vehicle already sitting at a station when + you subscribe did not just arrive and gets no `vehicle_arrive`. Its eventual `dwell` is counted + from that first sample, not from its real arrival. +- **`delta` is measured against the previous sample, not the previous event.** With a + `min_cargo_delta` filter, suppressed changes still move the baseline, so the next reported + `delta` covers everything since the last *sample*. + +Two more edge cases in how "at a stop" is defined: it means *loading or unloading* at a station, +so a vehicle passing through a station non-stop never registers, and a vehicle that is stopped by +hand or breaks down at a platform reads as having left (and as arriving again when it resumes). + +### Forwarded engine events — exact, as they happen + +These the engine does raise, so they are reported the moment they occur, with no sampling +involved and no `interval` dependence. + +| Kind | Fields | +|---|---| +| `vehicle_crashed` | `vehicle_id`, `tile`, `reason`, `victims`, `owner` | +| `station_first_vehicle` | `station_id`, `vehicle_id` | +| `industry_open`, `industry_close` | `industry_id` | +| `town_founded` | `town_id` | +| `company_new`, `company_in_trouble`, `company_bankrupt` | `company_id` | +| `subsidy_offer`, `subsidy_offer_expired`, `subsidy_awarded`, `subsidy_expired` | `subsidy_id` | + +Every event of either flavour also carries `event` (its kind) and `tick` (the game tick it was +observed at). + +There is deliberately no `vehicle_lost`, `vehicle_waiting_in_depot` or `vehicle_unprofitable`: +the engine raises those only for AI companies, never for a GameScript, so no bridge could +forward them. Watch `vehicle_arrive`/`vehicle_depart` or poll `get_timetable()` instead. + +### `events_dropped` + +Not something you subscribe to — the bridge emits it when a single poll produced more events than +its per-poll cap (200) and discarded `count` of them. Treat it as "your view has a hole in it": +if you see it, your subscription is too broad or your interval too coarse for what the map is +doing. Re-read state with `get_station()` / `get_timetable()` rather than trusting your +event-derived view. + +## `subscribe_events(...)` + +Awaitable; returns the bridge's confirmation `{"events": [accepted kinds], "interval": N}`. Every +argument narrows what you get, and all are optional. + +| Parameter | Type | Meaning | +|---|---|---| +| `events` | iterable | Which kinds to receive. Default: all of them. | +| `interval` | `int` | Ticks between state samples for the polled kinds. Default `10`. This is their resolution, not a delay. | +| `company_id` | `int` | Only report vehicles and stations owned by this company. | +| `vehicles` | iterable | Only sample these vehicle ids instead of every vehicle. | +| `stations` | iterable | Only sample these station ids instead of every station. | +| `cargo` | iterable | Only inspect these cargo types (for `cargo_waiting`, and for the load on vehicle events). | +| `min_cargo_delta` | `int` | Suppress `cargo_waiting` events smaller than this many units. Default `1` (report everything). | +| `include_cargo` | `bool` | `False` leaves the per-cargo load off vehicle events. Default `True`. | +| `timeout` | `float` | Seconds to wait for the confirmation. Default `5.0`. | + +Subscribing **replaces** any previous subscription and resets the baseline. `unsubscribe_events()` +stops the stream and drops the bridge's sampling state. + +Errors: `ValueError` when the GameScript rejects the request (`unknown_event`, +`invalid_interval`, `invalid_min_cargo_delta`, `invalid_cargo`), `asyncio.TimeoutError` when it +does not answer (GameScripts don't run while the game is **paused**, so a paused server always +times out), `ConnectionError` if the connection drops. The Gamescript update-frequency +subscription the stream needs is set up automatically. + +## Cost: subscribe narrowly + +The bridge samples every watched vehicle and every watched station × cargo pair on each interval, +inside a GameScript's limited per-tick opcode budget. The defaults — every kind, every vehicle, +every station, every cargo, every 10 ticks — are fine on a small map and wasteful on a large busy +one, where the poll can take longer than the interval it is trying to keep. + +Narrow it to what you actually need: + +```python +# Watch two specific vehicles closely. +await admin.subscribe_events( + events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart], + vehicles=[7, 9], interval=2) + +# Watch coal piling up at one station, ignoring changes under 10 units. +await admin.subscribe_events( + events=[GameEventType.CargoWaiting], + stations=[3], cargo=[0], min_cargo_delta=10, interval=30) + +# Only one company's traffic, without the per-vehicle cargo detail. +await admin.subscribe_events(company_id=0, include_cargo=False) +``` + +## Putting it together + +```python +import asyncio +from openttd import OpenTTDAdminClient +from openttd.protocol import GameEventType + +async def watch_route(admin, vehicle_id): + """Report how long a vehicle spends at each stop on its route.""" + await admin.subscribe_events( + events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart], + vehicles=[vehicle_id], interval=2) + try: + while True: + event = await admin.wait_for_event(timeout=300.0) + if event["event"] == "vehicle_arrive": + print(f"arrived at station {event['station_id']} (order {event['order_position']})") + else: + loaded = sum(c["load"] for c in event["cargo"]) + print(f"left station {event['station_id']} after {event['dwell']} ticks " + f"carrying {loaded} units") + except asyncio.TimeoutError: + print("vehicle went quiet") + finally: + await admin.unsubscribe_events() + +async def main(): + admin = OpenTTDAdminClient("127.0.0.1", admin_name="RouteWatcher") + await admin.connect(admin_password="asd", secure=True) + await admin.joined.wait() + await watch_route(admin, vehicle_id=7) + await admin.quit() + +asyncio.run(main()) +``` + +## Requirements + +The server must run the bundled AdminBridge GameScript (version 3 or newer). Events need **no** +server patch — unlike `get_timetable()` and `get_dispatch()`, every getter involved is part of the +stock GameScript API. Since GameScripts do not tick while the game is paused, no events are +produced on a paused server. + +## See also +- [PROTOCOL.md — Game Events](PROTOCOL.md#game-events) for the request/reply JSON. +- [TIMETABLES.md](TIMETABLES.md) for reading and editing what the events tell you changed. +- [ARCHITECTURE.md](ARCHITECTURE.md) for how the admin client fits together. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index c71358d..1ebd993 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -69,6 +69,16 @@ Correlation, the `update_frequency` subscription requirement (auto-subscribed on - **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. +### Game Events +Every other GameScript command above is request/reply. `subscribe_events()` instead opens a **push** stream: the AdminBridge GameScript sends event batches over `ServerGamescript` as things happen, unsolicited and without a `request_id`. For the calling API and the semantics of each kind, see the [Game Events Usage Guide](EVENTS.md). + +- **Subscribe request:** `{"command": "subscribe_events", "request_id": X}` plus any of the optional narrowing fields `"events"` (array of kind strings), `"interval"` (ticks between state samples, default 10), `"company_id"`, `"vehicles"`, `"stations"`, `"cargo"` (arrays of ids), `"min_cargo_delta"` (default 1) and `"include_cargo"` (bool, default true). +- **Subscribe reply:** `{"command": "subscribe_events", "request_id": X, "events": [accepted kinds in catalogue order], "interval": N}`, or the same envelope with an `"error"` field: `"unknown_event"` (plus the offending `"event"`), `"invalid_interval"`, `"invalid_min_cargo_delta"`, or `"invalid_cargo"` (plus the offending `"cargo_id"`). `subscribe_events()` raises `ValueError` for these. Subscribing replaces any previous subscription and resets the bridge's sampling baseline. +- **Unsubscribe:** `{"command": "unsubscribe_events", "request_id": X}` → `{"command": "unsubscribe_events", "request_id": X, "events": []}`. This also drops the sampling state. +- **Event batch (unsolicited):** `{"command": "events", "events": [{"event": , "tick": T, ...}, ...]}`. Batches carry at most 24 events per packet, so one poll can produce several; a poll that generated more than 200 events is truncated and followed by a single `{"event": "events_dropped", "count": N}` entry. Because these batches carry no `request_id`, `receive_ServerGamescript` routes them to the event consumers (`on_event` / `wait_for_event()`) instead of the generic `on_gamescript` callback; every other GameScript payload reaches `on_gamescript` unchanged. The usual `update_frequency(Gamescript, Automatic)` subscription applies and is set up automatically by `subscribe_events()`. + +Event kinds come from two sources. `vehicle_arrive`, `vehicle_depart` and `cargo_waiting` are **synthesised** by the bridge, because the engine raises no GameScript event for them: every `interval` ticks it samples `GSVehicle.GetState`/`GetLocation` for the watched vehicles and `GSStation.GetCargoWaiting` for the watched station-cargo pairs, and emits an event per change against the previous sample (so a stop shorter than the interval is never reported, and the first sample after subscribing only sets a baseline). All the remaining kinds — `vehicle_crashed`, `station_first_vehicle`, `industry_open`/`industry_close`, `town_founded`, `company_new`/`company_in_trouble`/`company_bankrupt` and the four `subsidy_*` kinds — are engine events forwarded verbatim from `GSEventController`. Only those the engine actually raises for a **deity** script are available: `ET_VEHICLE_LOST`, `ET_VEHICLE_WAITING_IN_DEPOT` and `ET_VEHICLE_UNPROFITABLE` are `@api ai` only and can never reach a GameScript, so the bridge does not offer them. Unlike the timetable and dispatch queries, none of this needs a server patch — every getter used is stock GameScript API. + ## 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). diff --git a/docs/TESTING.md b/docs/TESTING.md index d6366d4..b1bd2d5 100644 --- a/docs/TESTING.md +++ b/docs/TESTING.md @@ -13,6 +13,7 @@ The tests are located in the `tests/` directory: | [`test_admin.py`](file:///home/kovagoadi/openttd-client/tests/test_admin.py) | `OpenTTDAdminClient` | Tests admin client initialization, admin packet types, and basic protocol constants. | | [`test_protocol.py`](file:///home/kovagoadi/openttd-client/tests/test_protocol.py) | `OpenTTDProtocol` | Tests binary serialization, custom parsers, and stream encryption/decryption (XChaCha20-Poly1305). | | [`test_logic.py`](file:///home/kovagoadi/openttd-client/tests/test_logic.py) | `OpenTTDClient` | Tests client connection lifecycle, company joining flow, authentication, and state management. | +| [`test_events.py`](file:///home/kovagoadi/openttd-client/tests/test_events.py) | Game Events | Tests event subscription encoding, the push/pull consumption paths (`on_event`, `wait_for_event()`), buffering and waiter lifecycle. | | [`test_coverage.py`](file:///home/kovagoadi/openttd-client/tests/test_coverage.py) | Coverage Helpers | Auxiliary unit tests targeting connection errors, fallback packet handlers, and missing passwords to ensure high test coverage. | | [`test_e2e.py`](file:///home/kovagoadi/openttd-client/tests/test_e2e.py) | Integration / E2E | Connects to a running local OpenTTD server (e.g., in Docker) to verify full socket interactions, stream cryptography, and keep-alive frames. | diff --git a/lib/openttd/client.py b/lib/openttd/client.py index 758ec44..4e09140 100644 --- a/lib/openttd/client.py +++ b/lib/openttd/client.py @@ -3,6 +3,7 @@ import hashlib import logging import os import uuid +from collections import deque from typing import ClassVar import monocypher @@ -470,17 +471,17 @@ class OpenTTDClient: class OpenTTDAdminClient: """High-level OpenTTD Admin client.""" - def __init__(self, host, port=3977, admin_name="GeminiAdmin"): + def __init__(self, host, port=3977, admin_name="GeminiAdmin", event_buffer_size=256): self.host = host self.port = port self.admin_name = admin_name self.log = logging.getLogger(f"OTTDA-{admin_name}") - + # State self.encryption_enabled = False self.joined = asyncio.Event() self.shutdown_event = asyncio.Event() - + # Internal crypto self._admin_password = "" self._session_key_send = None @@ -488,17 +489,24 @@ class OpenTTDAdminClient: self._encryption_nonce = None self._send_aead = None self._recv_aead = None - + # Callbacks self.on_chat = None self.on_console = None self.on_gamescript = None + self.on_event = None # GameScript request/response correlation self._gs_request_id = 0 self._gs_futures = {} self._gs_subscribed = False + # Game events pushed by the AdminBridge GameScript. Events nobody is waiting for are + # kept here so a wait_for_event() call can still pick up something that arrived just + # before it; the deque bounds the memory a subscription nobody drains can cost. + self._event_buffer = deque(maxlen=event_buffer_size) + self._event_waiters = [] + async def connect(self, admin_password="", secure=False): """Connect to the admin port and initiate handshake.""" self._admin_password = admin_password @@ -532,6 +540,10 @@ class OpenTTDAdminClient: if not fut.done(): fut.set_exception(ConnectionError("admin disconnected")) self._gs_futures.clear() + for _, fut in self._event_waiters: + if not fut.done(): + fut.set_exception(ConnectionError("admin disconnected")) + self._event_waiters.clear() self.shutdown_event.set() async def quit(self): @@ -750,6 +762,131 @@ class OpenTTDAdminClient: {"command": "get_dispatch", "vehicle_id": vehicle_id}, timeout, f"get_dispatch({vehicle_id})") + # --- Game events --- + # + # Everything above is a request the caller makes; this is the other direction. The + # AdminBridge GameScript pushes events as they happen, so a bot can react to the game + # instead of polling it. Two kinds of thing arrive on the same channel: transitions the + # bridge synthesises by sampling game state on an interval (a vehicle reaching or leaving + # a stop, a station's waiting cargo changing), and the events the engine itself raises for + # a GameScript (crashes, industries opening, companies going bankrupt, ...). See + # GameEventType for the full catalogue. + # + # Consume them either by setting on_event (push) or by awaiting wait_for_event() (pull); + # both see every event, so the two can be mixed. + + async def subscribe_events(self, events=None, interval=None, company_id=None, vehicles=None, + stations=None, cargo=None, min_cargo_delta=None, + include_cargo=None, timeout=5.0): + """Ask the AdminBridge GameScript to start pushing game events, and wait for it to confirm. + + Every argument narrows what gets sent; the defaults subscribe to every event kind for + every vehicle, station and cargo, which is the right starting point on a small map and + the wrong one on a large busy map (see the cost note below). + + - events: which GameEventType kinds to receive (default: all of them). + - interval: ticks between state samples for the polled kinds (default 10). This is + the resolution of those events, not a delay: a stop shorter than `interval` can + begin and end between two samples and is then never reported at all. + - company_id: only report vehicles and stations owned by this company. + - vehicles / stations: only sample these ids, instead of every vehicle / station. + - cargo: only inspect these cargo types (for cargo_waiting and for the load reported + on vehicle events). + - min_cargo_delta: suppress cargo_waiting events whose amount moved by less than this + many units since the previous sample (default 1, i.e. report every change). + - include_cargo: set False to leave the per-cargo load off vehicle events. + + Subscribing replaces any previous subscription and resets the bridge's baseline, so the + first sample after this call only records where everything already is — a vehicle that + was sitting at a station when you subscribed did not just arrive, and gets no event. + + Cost: the bridge samples every watched vehicle and every watched station-cargo pair on + each interval, inside a GameScript's limited per-tick budget. On a large map prefer a + coarser interval and explicit vehicles/stations/cargo lists over the defaults. + + Returns the confirmation dict: {"events": [accepted kinds], "interval": N}. Raises + asyncio.TimeoutError if the GameScript does not answer (e.g. game paused, GS not + loaded), ValueError on a rejected request (unknown_event, invalid_interval, + invalid_min_cargo_delta, invalid_cargo), and ConnectionError if the admin connection + drops while waiting. + """ + payload = {"command": "subscribe_events"} + if events is not None: + payload["events"] = [str(event) for event in events] + if interval is not None: + payload["interval"] = interval + if company_id is not None: + payload["company_id"] = company_id + if vehicles is not None: + payload["vehicles"] = list(vehicles) + if stations is not None: + payload["stations"] = list(stations) + if cargo is not None: + payload["cargo"] = list(cargo) + if min_cargo_delta is not None: + payload["min_cargo_delta"] = min_cargo_delta + if include_cargo is not None: + payload["include_cargo"] = bool(include_cargo) + return await self._gs_query(payload, timeout, "subscribe_events") + + async def unsubscribe_events(self, timeout=5.0): + """Stop the event stream and wait for the GameScript to confirm. + + This also drops the bridge's sampling state, so a later subscribe_events() starts from + a fresh baseline. Events already delivered stay in this client's buffer; drain or ignore + them as you like. + """ + return await self._gs_query({"command": "unsubscribe_events"}, timeout, + "unsubscribe_events") + + async def wait_for_event(self, kind=None, timeout=5.0): + """Await the next game event, optionally of a specific kind (or any of several kinds). + + `kind` is a GameEventType (or plain string), an iterable of them, or None for "any + event". Events that arrived earlier and were not taken by another waiter are buffered, + so this returns immediately when a matching one is already in hand; the oldest matching + event wins. An event is handed to at most one waiter, but the on_event callback (if set) + still sees every event regardless. + + Returns the event dict, which always carries "event" (its GameEventType) and "tick" + (the game tick it was observed at) plus per-kind fields — see docs/EVENTS.md. Raises + asyncio.TimeoutError if nothing matching arrives in time (note that a subscription is + needed first — see subscribe_events()) and ConnectionError if the admin connection drops + while waiting. + """ + kinds = None + if kind is not None: + kinds = {str(kind)} if isinstance(kind, str) else {str(k) for k in kind} + for buffered in list(self._event_buffer): + if self._event_matches(buffered, kinds): + self._event_buffer.remove(buffered) + return buffered + fut = asyncio.get_running_loop().create_future() + waiter = (kinds, fut) + self._event_waiters.append(waiter) + try: + return await asyncio.wait_for(fut, timeout) + finally: + if waiter in self._event_waiters: + self._event_waiters.remove(waiter) + + @staticmethod + def _event_matches(event, kinds): + return kinds is None or (isinstance(event, dict) and event.get("event") in kinds) + + def _dispatch_event(self, event): + """Hand one event to the longest-waiting matching waiter, else buffer it; then observe.""" + for waiter in self._event_waiters: + kinds, fut = waiter + if not fut.done() and self._event_matches(event, kinds): + fut.set_result(event) + self._event_waiters.remove(waiter) + break + else: + self._event_buffer.append(event) + if self.on_event: + self.on_event(event) + async def send_gamescript(self, json_data): """Send a JSON string to the GameScript.""" import json @@ -852,6 +989,12 @@ class OpenTTDAdminClient: if not fut.done(): fut.set_result(data) return + # Unsolicited event batch from the AdminBridge GameScript: fan it out to the + # event consumers rather than the generic GameScript callback. + if data.get('command') == 'events' and isinstance(data.get('events'), list): + for event in data['events']: + self._dispatch_event(event) + return if self.on_gamescript: self.on_gamescript(data) else: diff --git a/lib/openttd/protocol.py b/lib/openttd/protocol.py index 9b98aa0..79c1b0b 100644 --- a/lib/openttd/protocol.py +++ b/lib/openttd/protocol.py @@ -1,5 +1,5 @@ import struct -from enum import IntEnum +from enum import IntEnum, StrEnum import monocypher from openttd_protocol.wire.exceptions import SocketClosed @@ -220,6 +220,40 @@ class NetworkAuthenticationMethod(IntEnum): X25519_PAKE = 1 X25519_AuthorizedKey = 2 +class GameEventType(StrEnum): + """Event kinds the AdminBridge GameScript can push over the Admin Network. + + These are the values of the "event" field of each event dict, and what + OpenTTDAdminClient.subscribe_events() and wait_for_event() take. They are plain strings, + so a bare "vehicle_arrive" works everywhere a member does. + + The first three are synthesised by the GameScript sampling game state on an interval, + because the engine raises no event for them; the rest are engine events forwarded as they + happen. VehicleLost, VehicleWaitingInDepot and VehicleUnprofitable have deliberately no + entry here: the engine only ever raises those for AI companies, never for a GameScript. + """ + + # Polled: derived by diffing successive samples of the game state. + VehicleArrive = "vehicle_arrive" + VehicleDepart = "vehicle_depart" + CargoWaiting = "cargo_waiting" + # Forwarded straight from the engine's own GameScript events. + VehicleCrashed = "vehicle_crashed" + StationFirstVehicle = "station_first_vehicle" + IndustryOpen = "industry_open" + IndustryClose = "industry_close" + TownFounded = "town_founded" + CompanyNew = "company_new" + CompanyInTrouble = "company_in_trouble" + CompanyBankrupt = "company_bankrupt" + SubsidyOffer = "subsidy_offer" + SubsidyOfferExpired = "subsidy_offer_expired" + SubsidyAwarded = "subsidy_awarded" + SubsidyExpired = "subsidy_expired" + # Emitted by the bridge itself, never subscribed to: one poll produced more events than + # fit in the per-poll cap and `count` of them were discarded. + EventsDropped = "events_dropped" + class OpenTTDProtocol(TCPProtocol): """Low-level OpenTTD TCP protocol handler with encryption support.""" PacketType = PacketGameType diff --git a/main_admin.py b/main_admin.py index baeccc0..4785a66 100644 --- a/main_admin.py +++ b/main_admin.py @@ -7,7 +7,7 @@ import sys sys.path.append(os.path.join(os.path.dirname(__file__), 'lib')) from openttd import OpenTTDAdminClient -from openttd.protocol import AdminUpdateFrequency, AdminUpdateType +from openttd.protocol import AdminUpdateFrequency, AdminUpdateType, GameEventType # Configuration SERVER_HOST = "127.0.0.1" @@ -85,7 +85,26 @@ async def run_admin(): except Exception as e: # noqa: BLE001 - demo script: one failed station query should not abort the walk print(f"!!! station query failed: {e}") - await asyncio.sleep(5) + # Watch the game live: vehicles reaching/leaving stops and cargo piling up at stations. + print("--- Subscribing to game events ---") + try: + accepted = await admin.subscribe_events( + events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart, + GameEventType.CargoWaiting], + interval=5, timeout=10.0) + print(f" subscribed to {accepted['events']} every {accepted['interval']} ticks") + + for _ in range(5): + try: + event = await admin.wait_for_event(timeout=15.0) + except asyncio.TimeoutError: + print(" (nothing happened -- is the server paused or idle?)") + break + print(f">>> [EVENT] {event}") + await admin.unsubscribe_events() + except Exception as e: # noqa: BLE001 - demo script: report and carry on to a clean quit + print(f"!!! event subscription failed: {e}") + print("--- Quitting ---") await admin.quit() diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 1c20645..437ae04 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -13,6 +13,7 @@ from openttd import OpenTTDAdminClient, OpenTTDClient from openttd.protocol import ( AdminUpdateFrequency, AdminUpdateType, + GameEventType, ModifyTimetableFlags, OpenTTDAdminProtocol, OpenTTDProtocol, @@ -27,6 +28,12 @@ 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 +# A second vehicle of the same company, dedicated to the scheduled-dispatch test, which owns and +# overwrites this vehicle's dispatch state. It must have its own order list -- dispatch schedules +# live on the order list, so pointing this at a vehicle that *shares* orders with another would +# silently rewrite that other vehicle's schedules too. (Cloning a vehicle without sharing orders +# gives an independent list, but copies the source's schedules along with it.) +DISPATCH_VEHICLE_ID = 14 # --- Pytest Fixtures --- @@ -303,17 +310,17 @@ async def test_e2e_client_scheduled_dispatch_edit_and_view(connected_owner_clien # 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 + veh = DISPATCH_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) + # The assertions below address schedules by absolute index, so DISPATCH_VEHICLE_ID must start + # with none of its own; the test restores that state on the way out. 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) @@ -716,6 +723,92 @@ async def test_e2e_admin_get_station_cargo_invalid_cargo(connected_admin): await connected_admin.get_station_cargo(stations[0]["id"], 250, timeout=10.0) +# --- Game Events --- + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_subscribe_events_all_kinds(connected_admin): + # Public functions: subscribe_events(), unsubscribe_events() + # Input 1: no arguments -> every event kind, default interval + data = await connected_admin.subscribe_events(timeout=10.0) + assert isinstance(data["events"], list) + assert "vehicle_arrive" in data["events"] and "cargo_waiting" in data["events"] + assert data["interval"] == 10 + + stopped = await connected_admin.unsubscribe_events() + assert stopped["events"] == [] + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_subscribe_events_filtered(connected_admin): + # Public functions: subscribe_events(), unsubscribe_events() + # Input 2: a narrowed subscription -> only the requested kinds come back + data = await connected_admin.subscribe_events( + events=[GameEventType.VehicleArrive, GameEventType.CargoWaiting], + interval=5, company_id=0, min_cargo_delta=2, include_cargo=False, timeout=10.0) + assert sorted(data["events"]) == ["cargo_waiting", "vehicle_arrive"] + assert data["interval"] == 5 + + await connected_admin.unsubscribe_events(timeout=10.0) + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_subscribe_events_unknown_kind(connected_admin): + # Public function: subscribe_events() + # A kind the GameScript does not know -> it reports unknown_event + with pytest.raises(ValueError, match="unknown_event"): + await connected_admin.subscribe_events(events=["definitely_not_an_event"], timeout=10.0) + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_wait_for_event_vehicle_reaches_a_stop(connected_admin): + # Public function: wait_for_event() + # Input 1: filtered by kind. Needs traffic on the server, so a quiet map skips. + await connected_admin.subscribe_events( + events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart], + interval=2, timeout=10.0) + try: + event = await connected_admin.wait_for_event( + {GameEventType.VehicleArrive, GameEventType.VehicleDepart}, timeout=60.0) + except asyncio.TimeoutError: + pytest.skip("No vehicle reached or left a stop on the test server within the timeout.") + finally: + await connected_admin.unsubscribe_events() + + assert event["event"] in ("vehicle_arrive", "vehicle_depart") + for key in ("tick", "vehicle_id", "station_id", "owner", "vehicle_type", "order_position"): + assert key in event + if event["event"] == "vehicle_depart": + assert event["dwell"] >= 0 + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_wait_for_event_any_kind(connected_admin): + # Public function: wait_for_event() + # Input 2: no kind filter -> whatever the game produces first + await connected_admin.subscribe_events(interval=2, timeout=10.0) + try: + event = await connected_admin.wait_for_event(timeout=60.0) + except asyncio.TimeoutError: + pytest.skip("Nothing happened on the test server within the timeout.") + finally: + await connected_admin.unsubscribe_events() + + assert "event" in event and "tick" in event + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_unsubscribe_events_stops_the_stream(connected_admin): + # Public function: unsubscribe_events() + # After unsubscribing the bridge must go quiet, so a fresh wait times out. + await connected_admin.subscribe_events(interval=2, timeout=10.0) + await connected_admin.unsubscribe_events(timeout=10.0) + + connected_admin._event_buffer.clear() # drop anything delivered before we unsubscribed + with pytest.raises(asyncio.TimeoutError): + await connected_admin.wait_for_event(timeout=5.0) + + # --- Protocol Public Functions --- @pytest.mark.e2e diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..70a378e --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,251 @@ +import asyncio +import json + +import pytest +from openttd import OpenTTDAdminClient +from openttd.protocol import GameEventType, PacketAdminType + + +class MockTransport: + def is_closing(self): + return False + def close(self): + pass + 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_gamescript_payload(packet): + """Decode the JSON payload of an AdminGamescript packet (2-byte length + 1-byte type + string).""" + return json.loads(packet[3:].split(b"\x00")[0]) + +def make_admin(subscribed=True, **kwargs): + """An admin client wired to mock transports, with the Gamescript auto-subscribe already done + unless a test wants to observe it.""" + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin", **kwargs) + client._protocol = MockProtocol() + client._transport = MockTransport() + client._gs_subscribed = subscribed + return client + +def events_packet(*events): + """The envelope the AdminBridge GameScript pushes event batches in.""" + return {"command": "events", "events": list(events)} + +def arrival(vehicle_id=7, station_id=3, tick=100): + return {"event": "vehicle_arrive", "tick": tick, "vehicle_id": vehicle_id, + "station_id": station_id, "owner": 0, "vehicle_type": 0, "order_position": 0, + "cargo": [{"cargo_id": 0, "load": 12}]} + + +# --- Event kinds --- + +def test_event_type_values_are_the_wire_strings(): + assert GameEventType.VehicleArrive == "vehicle_arrive" + assert GameEventType.CargoWaiting == "cargo_waiting" + # Members are plain strings, so they can be used interchangeably with literals. + assert str(GameEventType.VehicleDepart) == "vehicle_depart" + + +# --- Subscribing --- + +@pytest.mark.asyncio +async def test_subscribe_events_defaults_send_only_the_command(): + client = make_admin(subscribed=False) + + task = asyncio.ensure_future(client.subscribe_events()) + await asyncio.sleep(0) + + # First use auto-subscribes to Gamescript updates, then sends the request. + assert len(client._protocol.sent) == 2 + assert client._protocol.sent[0][2] == PacketAdminType.AdminUpdateFrequency + assert decode_gamescript_payload(client._protocol.sent[1]) == { + "command": "subscribe_events", "request_id": 1, + } + + reply = {"command": "subscribe_events", "request_id": 1, + "events": ["vehicle_arrive", "vehicle_depart"], "interval": 10} + await client.receive_ServerGamescript(None, data=reply) + assert await task == reply + assert client._gs_futures == {} + +@pytest.mark.asyncio +async def test_subscribe_events_encodes_every_filter(): + client = make_admin() + + task = asyncio.ensure_future(client.subscribe_events( + events=[GameEventType.VehicleArrive, "cargo_waiting"], interval=25, company_id=0, + vehicles=(7, 9), stations=(3,), cargo=[0, 1], min_cargo_delta=5, include_cargo=False)) + await asyncio.sleep(0) + + assert decode_gamescript_payload(client._protocol.sent[0]) == { + "command": "subscribe_events", "request_id": 1, + "events": ["vehicle_arrive", "cargo_waiting"], "interval": 25, "company_id": 0, + "vehicles": [7, 9], "stations": [3], "cargo": [0, 1], + "min_cargo_delta": 5, "include_cargo": False, + } + await client.receive_ServerGamescript( + None, data={"request_id": 1, "events": ["vehicle_arrive", "cargo_waiting"]}) + assert (await task)["events"] == ["vehicle_arrive", "cargo_waiting"] + +@pytest.mark.asyncio +async def test_subscribe_events_rejected_request_raises(): + client = make_admin() + + task = asyncio.ensure_future(client.subscribe_events(events=["not_an_event"])) + await asyncio.sleep(0) + await client.receive_ServerGamescript( + None, data={"command": "subscribe_events", "request_id": 1, + "error": "unknown_event", "event": "not_an_event"}) + with pytest.raises(ValueError, match="unknown_event"): + await task + assert client._gs_futures == {} + +@pytest.mark.asyncio +async def test_subscribe_events_timeout(): + client = make_admin() + with pytest.raises(asyncio.TimeoutError): + await client.subscribe_events(timeout=0.05) + assert client._gs_futures == {} + +@pytest.mark.asyncio +async def test_unsubscribe_events(): + client = make_admin() + + task = asyncio.ensure_future(client.unsubscribe_events()) + await asyncio.sleep(0) + assert decode_gamescript_payload(client._protocol.sent[0]) == { + "command": "unsubscribe_events", "request_id": 1, + } + + reply = {"command": "unsubscribe_events", "request_id": 1, "events": []} + await client.receive_ServerGamescript(None, data=reply) + assert await task == reply + + +# --- Receiving events --- + +@pytest.mark.asyncio +async def test_event_batch_reaches_on_event_and_bypasses_on_gamescript(): + client = make_admin() + seen, other = [], [] + client.on_event = seen.append + client.on_gamescript = other.append + + depart = {"event": "vehicle_depart", "tick": 120, "vehicle_id": 7, + "station_id": 3, "dwell": 74} + await client.receive_ServerGamescript(None, data=events_packet(arrival(), depart)) + + assert [event["event"] for event in seen] == ["vehicle_arrive", "vehicle_depart"] + assert other == [] + +@pytest.mark.asyncio +async def test_non_event_gamescript_payloads_still_reach_on_gamescript(): + client = make_admin() + seen, other = [], [] + client.on_event = seen.append + client.on_gamescript = other.append + + # A reply that merely mentions events, and a malformed batch, are not event batches. + await client.receive_ServerGamescript(None, data={"vehicles": []}) + await client.receive_ServerGamescript(None, data={"command": "events", "events": "nope"}) + + assert seen == [] + assert other == [{"vehicles": []}, {"command": "events", "events": "nope"}] + +@pytest.mark.asyncio +async def test_events_are_buffered_until_awaited(): + client = make_admin() + await client.receive_ServerGamescript(None, data=events_packet(arrival(station_id=3), + arrival(station_id=4))) + + # The oldest matching event is handed out first, and only once. + first = await client.wait_for_event("vehicle_arrive") + second = await client.wait_for_event(GameEventType.VehicleArrive) + assert (first["station_id"], second["station_id"]) == (3, 4) + assert len(client._event_buffer) == 0 + +@pytest.mark.asyncio +async def test_wait_for_event_resolves_on_arrival(): + client = make_admin() + task = asyncio.ensure_future(client.wait_for_event("cargo_waiting", timeout=5.0)) + await asyncio.sleep(0) + + cargo = {"event": "cargo_waiting", "tick": 300, "station_id": 3, + "cargo_id": 0, "waiting": 25, "delta": 15} + await client.receive_ServerGamescript(None, data=events_packet(arrival(), cargo)) + + assert await task == cargo + # The event that did not match the waiter is still buffered for the next caller. + assert await client.wait_for_event() == arrival() + assert client._event_waiters == [] + +@pytest.mark.asyncio +async def test_wait_for_event_accepts_several_kinds(): + client = make_admin() + task = asyncio.ensure_future( + client.wait_for_event({GameEventType.VehicleArrive, GameEventType.VehicleDepart})) + await asyncio.sleep(0) + + await client.receive_ServerGamescript( + None, data=events_packet({"event": "town_founded", "tick": 5, "town_id": 2}, arrival())) + assert (await task)["event"] == "vehicle_arrive" + +@pytest.mark.asyncio +async def test_wait_for_event_ignores_non_dict_events_when_filtering(): + client = make_admin() + task = asyncio.ensure_future(client.wait_for_event("vehicle_arrive", timeout=0.05)) + await asyncio.sleep(0) + + # A malformed entry inside a batch must not satisfy a filtered waiter. + await client.receive_ServerGamescript(None, data=events_packet("garbage")) + with pytest.raises(asyncio.TimeoutError): + await task + assert list(client._event_buffer) == ["garbage"] + +@pytest.mark.asyncio +async def test_wait_for_event_timeout_deregisters_the_waiter(): + client = make_admin() + with pytest.raises(asyncio.TimeoutError): + await client.wait_for_event("vehicle_arrive", timeout=0.05) + assert client._event_waiters == [] + +@pytest.mark.asyncio +async def test_already_resolved_waiters_are_skipped(): + client = make_admin() + # A waiter whose future completed but that has not been cleaned up yet must not swallow + # the event; it goes to the next live waiter instead. + stale = asyncio.get_running_loop().create_future() + stale.set_result("already done") + live = asyncio.get_running_loop().create_future() + client._event_waiters = [(None, stale), (None, live)] + + await client.receive_ServerGamescript(None, data=events_packet(arrival())) + assert live.result() == arrival() + assert len(client._event_buffer) == 0 + +@pytest.mark.asyncio +async def test_event_buffer_is_bounded(): + client = make_admin(event_buffer_size=2) + await client.receive_ServerGamescript(None, data=events_packet( + arrival(station_id=1), arrival(station_id=2), arrival(station_id=3))) + + # The oldest event is dropped rather than growing the buffer without bound. + assert [event["station_id"] for event in client._event_buffer] == [2, 3] + +@pytest.mark.asyncio +async def test_disconnect_fails_pending_event_waiters(): + client = make_admin() + task = asyncio.ensure_future(client.wait_for_event()) + await asyncio.sleep(0) + + client.disconnect(None) + with pytest.raises(ConnectionError): + await task + assert client._event_waiters == [] -- 2.54.0