TfL Unified API - Is a Tube Line Running Normally, and the Next Three Trains at a Station (Transport for London open data, official API)

import json, time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from zoneinfo import ZoneInfo
 
LINE, STATION = "victoria", "Oxford Circus"  # line id as the API spells it; station name to look up
 
SITE = "https://api.tfl.gov.uk"
PAUSE = 1.0  # TfL licence: "maximum of 500 calls per minute per data feed"; keep calls serial
LONDON = ZoneInfo("Europe/London")
ATTRIBUTION = "Powered by TfL Open Data. Contains OS data © Crown copyright and database rights 2016 and Geomni UK Map data © and database rights [2019]."  # licence: both statements
 
def on_api_tab():
    """fetch() runs same-origin from a logged-out api.tfl.gov.uk tab (the line's own status JSON). Reuse it."""
    try:
        if js("location.origin") == SITE:
            return
    except Exception:
        pass
    new_tab(f"{SITE}/Line/{LINE}/Status")
    wait_for_load()
    body = js("document.body ? document.body.innerText : ''") or ""
    if js("location.origin") != SITE or '"lineStatuses"' not in body:
        raise RuntimeError(f"api.tfl.gov.uk did not serve the line status JSON (at {js('location.href')}): stop")
 
# js() gives up after a few seconds, so start the fetch, then poll for it.
START_JS = """(() => { window.__tfl = window.__tfl || {}; const id = String(Math.random()).slice(2); window.__tfl[id] = null;
  fetch(%s, {credentials: 'omit'})
    .then(async r => { window.__tfl[id] = {status: r.status, type: r.headers.get('content-type'), date: r.headers.get('date'), text: await r.text()}; })
    .catch(e => { window.__tfl[id] = {status: 0, type: '', date: null, text: String(e)}; });
  return id; })()"""
POLL_JS = """(() => { const v = window.__tfl[%s]; if (v) delete window.__tfl[%s]; return v; })()"""
 
def get(path, timeout=30):
    """GET SITE+path from inside the tab -> parsed JSON. Errors are JSON ApiError {"httpStatusCode":404,"message":...}.
    Waits 30 s once on 429/5xx, raises on anything else, so an error never looks like an answer."""
    for attempt in range(2):
        rid = json.dumps(js(START_JS % json.dumps(SITE + path)))
        t0 = time.time()
        while (r := js(POLL_JS % (rid, rid))) is None:
            if time.time() - t0 > timeout:
                raise RuntimeError(f"{path}: no answer after {timeout}s")
            time.sleep(0.3)
        time.sleep(PAUSE)
        if r["status"] == 200 and "json" in (r["type"] or ""):
            get.date = r["date"]  # server clock (HTTP Date header); used as "now" so a skewed local clock does not matter
            return json.loads(r["text"])
        if r["status"] in (0, 429, 500, 502, 503, 504) and attempt == 0:
            time.sleep(30)
            continue
        raise RuntimeError(f"{path}: HTTP {r['status']} {' '.join(r['text'].split())[:200]!r}; stopping")
 
def ts(s):
    """TfL times are UTC ISO strings with up to 7 fractional digits ('2026-11-12T23:27:28.3735786Z')."""
    s = s.rstrip("Z")
    if "." in s:
        head, frac = s.split(".")
        s = head + "." + frac[:6]
    return datetime.fromisoformat(s).replace(tzinfo=timezone.utc)
 
on_api_tab()
 
# 1. Station id: search, keep the tube match whose name starts with the station name (940GZZLUOXC for Oxford Circus).
hits = get("/StopPoint/Search/" + STATION.replace(" ", "%20") + "?modes=tube")["matches"]
hits = [m for m in hits if m["name"].lower().startswith(STATION.lower())]
if len(hits) != 1:
    raise RuntimeError(f"expected one tube station for {STATION!r}, got {[m['name'] for m in hits]}")
naptan = hits[0]["id"]
print(f"station: {hits[0]['name']} -> {naptan}")
# A station the line does not serve gives HTTP 200 [] arrivals, which would look like "no trains": rule it out first.
if naptan not in {s["naptanId"] for s in get(f"/Line/{LINE}/StopPoints")}:
    raise RuntimeError(f"{hits[0]['name']} is not on the {LINE} line: stop")
 
# 2. Line status. Severity 10 = "Good Service" (/Line/Meta/Severity); anything else comes with a reason.
line = get(f"/Line/{LINE}/Status")
if len(line) != 1 or line[0]["id"] != LINE:
    raise RuntimeError(f"asked {LINE}, got {[l['id'] for l in line]}")
line = line[0]
statuses = line["lineStatuses"]
if not statuses:
    raise RuntimeError("no lineStatuses in the answer: stop")
normal = all(s["statusSeverity"] == 10 for s in statuses)
print(f"\n{line['name']} line: " + "; ".join(f"{s['statusSeverityDescription']} (severity {s['statusSeverity']})" for s in statuses))
for s in statuses:
    if s.get("reason"):
        print("  reason:", " ".join(s["reason"].split()))
 
# 3. Arrivals for this line at this station. One call; the server caches 30 s (cache-control max-age=30).
preds = get(f"/Line/{LINE}/Arrivals/{naptan}")
now = parsedate_to_datetime(get.date) if get.date else datetime.now(timezone.utc)
skew = (datetime.now(timezone.utc) - now).total_seconds()
if abs(skew) > 5:
    print(f"  note: local clock is {skew:+.0f} s off the server; using the server's Date header")
bad = [p for p in preds if p["lineId"] != LINE or p["naptanId"] != naptan]
if bad:
    raise RuntimeError(f"{len(bad)} predictions for another line or stop: stop")
if preds:
    age = (now - max(ts(p["timestamp"]) for p in preds)).total_seconds()
    print(f"\n{len(preds)} predictions, newest computed {age:.0f} s ago")
    if age > 120:
        raise RuntimeError(f"predictions are {age:.0f} s old; the feed is stale, do not quote them")
 
# timeToStation is seconds at prediction time; recompute from expectedArrival so a cached answer does not run fast.
best = {}
for p in preds:
    secs = (ts(p["expectedArrival"]) - now).total_seconds()
    if secs < 0:  # expected time has passed: the train is in the platform or gone
        continue
    key = (p["vehicleId"], p["platformName"])  # one train can appear twice; keep its earliest row per platform
    if key not in best or secs < best[key][0]:
        best[key] = (secs, p)
trains = sorted(best.values(), key=lambda t: t[0])
 
print(f"\nnow {now.astimezone(LONDON):%H:%M:%S %Z}")
by_platform = {}
for secs, p in trains:
    by_platform.setdefault(p["platformName"], []).append((secs, p))
if not by_platform:
    print(f"no {line['name']} line trains predicted at this station right now (it is on the line; see the status above)")
for plat in sorted(by_platform):
    print(f"{plat}:")
    for secs, p in by_platform[plat][:3]:
        when = ts(p["expectedArrival"]).astimezone(LONDON)
        mins = "due" if secs < 30 else f"{round(secs / 60)} min"
        print(f"  {when:%H:%M}  {mins:>6}  to {p['towards'] or p['destinationName']}  ({p['currentLocation']}, train {p['vehicleId']})")
 
nxt = trains[:3]
print(f"\nanswer: {line['name']} line {'running normally' if normal else 'NOT running normally'} "
      f"({', '.join(s['statusSeverityDescription'] for s in statuses)}).")
print("  next three at " + hits[0]["name"].replace(" Underground Station", "") + ": "
      + ("; ".join(f"{ts(p['expectedArrival']).astimezone(LONDON):%H:%M} {p['platformName'].split(' ')[0].lower()} to {p['towards'] or p['destinationName']}" for _, p in nxt) or "none predicted"))
print("  " + ATTRIBUTION)

station: Oxford Circus Underground Station -> 940GZZLUOXC
 
Victoria line: Good Service (severity 10)
 
3 predictions, newest computed 50 s ago
 
now 00:01:40 GMT
Southbound - Platform 5:
  00:04   3 min  to Brixton  (Between Euston and Warren Street, train 273)
  00:09   8 min  to Brixton  (Between Finsbury Park and Highbury & Islington, train 206)
  00:14  13 min  to Brixton  (At Seven Sisters Platform 5, train 202)
 
answer: Victoria line running normally (Good Service).
  next three at Oxford Circus: 00:04 southbound to Brixton; 00:09 southbound to Brixton; 00:14 southbound to Brixton
  Powered by TfL Open Data. Contains OS data © Crown copyright and database rights 2016 and Geomni UK Map data © and database rights [2019].

import json, time
 
SITE = "https://api.tfl.gov.uk"
TERMS = "https://tfl.gov.uk/corporate/terms-and-conditions/transport-data-service"
PORTAL = "https://api-portal.tfl.gov.uk/"
 
def page_text(url, must, timeout=30):
    """Load url and wait until `must` is in the text. tfl.gov.uk shows a Cloudflare check
    ("Just a moment...") for about 5 s in a fresh browser, then the page."""
    goto_url(url)
    wait_for_load()
    t0 = time.time()
    while True:
        t = " ".join((js("document.body ? document.body.innerText : ''") or "").split())
        if must in t:
            return t
        if time.time() - t0 > timeout:
            if "Verification required" in t or "Just a moment" in (js("document.title") or ""):
                raise RuntimeError(f"{url}: Cloudflare check did not clear in {timeout}s; read it later, do not guess the terms")
            raise RuntimeError(f"{must!r} not on {url}: the page changed, re-read it before relying on this skill")
        time.sleep(1)
 
def quote(text, start, n=300):
    i = text.find(start)
    if i < 0:
        raise RuntimeError(f"{start!r} not found: the page changed, re-read it before relying on this skill")
    return text[i:i + n]
 
# Licence terms, as the page shows them.
new_tab("about:blank")
t = page_text(TERMS, "You are free to:")
for s, n in (("You are free to:", 240), ("Acknowledge TfL as the source", 260), ("Limit traffic requests", 200),
             ("Use any automated system", 220), ("This Licence does not grant You any right to use the Information in a way", 160)):
    print(f"terms: {quote(t, s, n).rstrip()}\n")
for word in ("artificial intelligence", "machine learning", " AI ", "non-commercial use only"):
    print(f"terms mention {word.strip()!r}: {word.lower() in t.lower()}")
 
# The API portal on keys.
t = page_text(PORTAL, "Unified API")
i = t.lower().find("recommended")
print("\nportal:", t[max(0, i - 200):i + 200] if i >= 0 else "no 'recommended' on the home page")
 
# Same-origin probes from an api.tfl.gov.uk page. No retries: this block only inspects answers.
goto_url(SITE + "/Line/Meta/Severity")
wait_for_load()
if js("location.origin") != SITE:
    raise RuntimeError(f"not on {SITE} (at {js('location.href')}): stop")
 
START_JS = """(() => { window.__tfl = window.__tfl || {}; const id = String(Math.random()).slice(2); window.__tfl[id] = null;
  fetch(%s, {credentials: 'omit'})
    .then(async r => { window.__tfl[id] = {status: r.status, type: r.headers.get('content-type'), cache: r.headers.get('cache-control'), text: await r.text()}; })
    .catch(e => { window.__tfl[id] = {status: 0, type: '', cache: '', text: String(e)}; });
  return id; })()"""
POLL_JS = """(() => { const v = window.__tfl[%s]; if (v) delete window.__tfl[%s]; return v; })()"""
 
def raw(path, timeout=30):
    rid = json.dumps(js(START_JS % json.dumps(SITE + path)))
    t0 = time.time()
    while (r := js(POLL_JS % (rid, rid))) is None:
        if time.time() - t0 > timeout:
            raise RuntimeError(f"{path}: no answer after {timeout}s")
        time.sleep(0.3)
    time.sleep(1.0)
    return r
 
r = raw("/robots.txt")
print(f"\napi robots.txt: HTTP {r['status']} {r['type']} {json.loads(r['text'])['message']!r}")
 
sev = json.loads(raw("/Line/Meta/Severity")["text"])
print("tube severities:", ", ".join(f"{x['severityLevel']} {x['description']}" for x in sev if x["modeName"] == "tube"))
 
r = raw("/Line/Mode/tube/Status")
print(f"\n/Line/Mode/tube/Status: HTTP {r['status']} cache-control {r['cache']!r}")
for l in json.loads(r["text"]):
    for s in l["lineStatuses"]:
        why = " ".join((s.get("reason") or "").split())
        print(f"  {l['id']:<16} {s['statusSeverity']:>2} {s['statusSeverityDescription']:<14} {why[:110]}".rstrip())
 
for path in ("/Line/victoriaa/Status",          # typo in the line id
             "/StopPoint/940GZZLUXXX/Arrivals",  # unknown stop
             "/StopPoint/Search/zzqqxx",         # no match
             "/Line/victoria/Arrivals/940GZZLUBST"):  # a station the Victoria line does not serve (Baker Street)
    r = raw(path)
    d = json.loads(r["text"])
    body = (f"{d['exceptionType']}: {d['message']}" if isinstance(d, dict) and "httpStatusCode" in d
            else json.dumps({k: v for k, v in d.items() if k != "$type"}) if isinstance(d, dict) else r["text"])
    print(f"{path}: HTTP {r['status']} {body}")

terms: You are free to: Copy, publish, distribute and transmit the Information Adapt the Information and Exploit the Information commercially and non-commercially for example, by combining it with other Information, or by including it in Your own
 
terms: Acknowledge TfL as the source of the Information by including the following attribution statement ‘Powered by TfL Open Data’ Acknowledge that this Information contains Ordnance Survey derived data by including the following attribution statement: ‘Contains OS
 
terms: Limit traffic requests up to a maximum of 500 calls per minute per data feed. TfL reserves the right to throttle or limit access to feeds when it is believed the overall service is being degraded by e
 
terms: Use any automated system, software or process to extract content and/or data, including trawling, data mining and screen scraping in relation to the Oyster, Congestion Charging and Santander Cycles websites, except where
 
terms: This Licence does not grant You any right to use the Information in a way that suggests any official status or that TfL endorses You or Your use of the Informat
 
terms mention 'artificial intelligence': False
terms mention 'machine learning': False
terms mention 'AI': False
terms mention 'non-commercial use only': False
 
portal: Home APIs FAQ Sign in Register To use the Unified API, it is recommended to register for an account. During registration, you should receive a confirmation email containing a link to activate your account. If you have not received the email, please check your s
 
api robots.txt: HTTP 404 application/json; charset=utf-8 'Resource not found: http://api:8001/robots.txt'
tube severities: 0 Special Service, 1 Closed, 2 Suspended, 3 Part Suspended, 4 Planned Closure, 5 Part Closure, 6 Severe Delays, 7 Reduced Service, 8 Bus Service, 9 Minor Delays, 10 Good Service, 11 Part Closed, 12 Exit Only, 13 No Step Free Access, 14 Change of frequency, 15 Diverted, 16 Not Running, 17 Issues Reported, 18 No Issues, 19 Information, 20 Service Closed
 
/Line/Mode/tube/Status: HTTP 200 cache-control 'public, must-revalidate, max-age=30, s-maxage=60'
  bakerloo          9 Minor Delays   Bakerloo Line: Minor delays between Queen's park and Harrow & Wealdstone due to an earlier signal fault at Sou
  central          10 Good Service
  circle            9 Minor Delays   Circle Line: Minor delays due to train cancellations.
  district          9 Minor Delays   District Line: Minor delays due to an earlier faulty train at Whitechapel and train cancellations.
  hammersmith-city 10 Good Service
  jubilee           9 Minor Delays   Jubilee Line: Minor delays due to train cancellations.
  metropolitan     10 Good Service
  northern          6 Severe Delays  Northern Line: Severe delays due to an earlier points failure at Mornington Crescent. Tickets are being accept
  piccadilly       10 Good Service
  victoria         10 Good Service
  waterloo-city    10 Good Service
/Line/victoriaa/Status: HTTP 404 EntityNotFoundException: The following line id is not recognised: victoriaa
/StopPoint/940GZZLUXXX/Arrivals: HTTP 404 EntityNotFoundException: The following stop point is not recognised: 940GZZLUXXX
/StopPoint/Search/zzqqxx: HTTP 200 {"query": "zzqqxx", "total": 0, "matches": []}
/Line/victoria/Arrivals/940GZZLUBST: HTTP 200 []

import json, sys, time, urllib.error, urllib.request
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from zoneinfo import ZoneInfo
 
LINE, NAPTAN = "victoria", "940GZZLUOXC"  # Oxford Circus (from /StopPoint/Search/oxford%20circus?modes=tube)
API = "https://api.tfl.gov.uk"
LONDON = ZoneInfo("Europe/London")
ATTRIBUTION = "Powered by TfL Open Data. Contains OS data © Crown copyright and database rights 2016 and Geomni UK Map data © and database rights [2019]."
UA = {"User-Agent": "tube-status-example/1.0"}  # Python's default "Python-urllib/3.12" gets Cloudflare 403 "error code: 1010"
 
def get(path):
    """GET -> (status, Date header, body). No key: app_key is optional (the portal 'recommended' registering)."""
    try:
        with urllib.request.urlopen(urllib.request.Request(API + path, headers=UA), timeout=30) as r:
            return r.status, r.headers.get("Date"), r.read().decode()
    except urllib.error.HTTPError as e:
        return e.code, e.headers.get("Date"), e.read().decode()
 
def ts(s):
    s = s.rstrip("Z")
    if "." in s:
        head, frac = s.split(".")
        s = head + "." + frac[:6]  # TfL sends up to 7 fractional digits
    return datetime.fromisoformat(s).replace(tzinfo=timezone.utc)
 
s, _, body = get(f"/Line/{LINE}/Status")
if s != 200:
    sys.exit(f"status: HTTP {s} {body[:200]}")
statuses = json.loads(body)[0]["lineStatuses"]
normal = bool(statuses) and all(x["statusSeverity"] == 10 for x in statuses)
print("Victoria:", ", ".join(x["statusSeverityDescription"] for x in statuses), "->", "normal" if normal else "NOT normal")
for x in statuses:
    if x.get("reason"):
        print("  reason:", " ".join(x["reason"].split()))
 
time.sleep(1)
s, date, body = get(f"/Line/{LINE}/Arrivals/{NAPTAN}")
if s != 200:
    sys.exit(f"arrivals: HTTP {s} {body[:200]}")
now = parsedate_to_datetime(date)  # server clock
preds = json.loads(body)
if preds:  # responses are cached; refuse a stale feed as block 1 does
    age = (now - max(ts(p["timestamp"]) for p in preds)).total_seconds()
    if age > 120:
        sys.exit(f"predictions are {age:.0f} s old; the feed is stale, do not quote them")
trains = {}
for p in preds:
    if p["lineId"] != LINE or p["naptanId"] != NAPTAN:
        sys.exit("prediction for another line or stop: stop")
    secs = (ts(p["expectedArrival"]) - now).total_seconds()
    key = (p["vehicleId"], p["platformName"])  # one row per train per platform, its earliest
    if secs >= 0 and (key not in trains or secs < trains[key][0]):
        trains[key] = (secs, p)
nxt = sorted(trains.values(), key=lambda t: t[0])[:3]
if not nxt:
    sys.exit("no Victoria line trains predicted at Oxford Circus: check the line status")
print(f"now {now.astimezone(LONDON):%H:%M:%S %Z}, next three at Oxford Circus:")
for secs, p in nxt:
    print(f"  {ts(p['expectedArrival']).astimezone(LONDON):%H:%M}  {round(secs / 60):>2} min  {p['platformName']:<24} to {p['towards'] or p['destinationName']}")
print(ATTRIBUTION)

Victoria: Good Service -> normal
now 23:54:54 GMT, next three at Oxford Circus:
  00:00   5 min  Southbound - Platform 5  to Brixton
  00:04   9 min  Southbound - Platform 5  to Brixton
  00:11  16 min  Southbound - Platform 5  to Brixton
Powered by TfL Open Data. Contains OS data © Crown copyright and database rights 2016 and Geomni UK Map data © and database rights [2019].

A=https://api.tfl.gov.uk   # no key needed; add ?app_key=... only if you registered one
curl -sS --fail "$A/Line/victoria/Status" | python3 -c '
import json, sys
s = json.load(sys.stdin)[0]["lineStatuses"]
print("victoria:", [(x["statusSeverity"], x["statusSeverityDescription"]) for x in s])
'
sleep 1
curl -sS --fail "$A/Line/victoria/Arrivals/940GZZLUOXC" | python3 -c '
import json, sys
d = sorted(json.load(sys.stdin), key=lambda p: p["timeToStation"])
for p in d[:3]:
    print(p["timeToStation"], "s", p["platformName"], "->", p["towards"], p["expectedArrival"])
'
sleep 1
curl -sS -o /dev/null -w 'unknown line: HTTP %{http_code} %{content_type}\n' "$A/Line/victoriaa/Status"
echo 'Powered by TfL Open Data. Contains OS data © Crown copyright and database rights 2016 and Geomni UK Map data © and database rights [2019].'

victoria: [(10, 'Good Service')]
325 s Southbound - Platform 5 -> Brixton 2026-11-12T23:59:30Z
565 s Southbound - Platform 5 -> Brixton 2026-11-13T00:03:30Z
985 s Southbound - Platform 5 -> Brixton 2026-11-13T00:10:30Z
unknown line: HTTP 404 application/json; charset=utf-8
Powered by TfL Open Data. Contains OS data © Crown copyright and database rights 2016 and Geomni UK Map data © and database rights [2019].