diff --git a/frontend/src/test/LiveTimingPage.test.tsx b/frontend/src/test/LiveTimingPage.test.tsx
index f39af28..cb831e6 100644
--- a/frontend/src/test/LiveTimingPage.test.tsx
+++ b/frontend/src/test/LiveTimingPage.test.tsx
@@ -152,4 +152,15 @@ describe('LiveTimingPage archive mode', () => {
expect(await screen.findByTestId('live-empty')).toHaveTextContent('No live session active')
expect(screen.queryByRole('button', { name: /view last session/i })).not.toBeInTheDocument()
})
+
+ it('temporarily omits the track map while live GPS is unavailable', async () => {
+ renderPage({
+ is_live: true,
+ data: { ...archivedSnapshot, SessionStatus: 'Started' },
+ })
+
+ expect(await screen.findByText('Timing Tower')).toBeInTheDocument()
+ expect(screen.queryByText('Track Map')).not.toBeInTheDocument()
+ expect(mockFetchLiveTrackOutline).not.toHaveBeenCalled()
+ })
})
diff --git a/scripts/debug/f1_live_socket_probe.py b/scripts/debug/f1_live_socket_probe.py
new file mode 100644
index 0000000..022fad1
--- /dev/null
+++ b/scripts/debug/f1_live_socket_probe.py
@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+"""Probe the official F1 live timing sockets for GPS position topics.
+
+This intentionally does not call OpenF1 REST. It connects directly to the F1
+SignalR endpoints used by the app and reports which topics arrive.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import os
+import random
+import socket
+import ssl
+import struct
+import sys
+import time
+import urllib.parse
+import urllib.request
+
+
+RS = b"\x1e"
+CORE_NEGOTIATE = "https://livetiming.formula1.com/signalrcore/negotiate?negotiateVersion=1"
+CORE_WS_HOST = "livetiming.formula1.com"
+LEGACY_HUB = '[{"name":"Streaming"}]'
+TOPICS = [
+ "Heartbeat",
+ "TimingData",
+ "Position",
+ "Position.z",
+ "CarData",
+ "CarData.z",
+ "DriverList",
+ "LapCount",
+ "ExtrapolatedClock",
+ "TrackStatus",
+ "RaceControlMessages",
+ "WeatherData",
+ "SessionInfo",
+ "TeamRadio",
+ "CurrentTyres",
+ "TimingAppData",
+ "TimingStats",
+ "SessionStatus",
+ "TopThree",
+]
+
+
+class WebSocket:
+ def __init__(self, host: str, path: str, headers: dict[str, str] | None = None):
+ self.host = host
+ self.sock = ssl.create_default_context().wrap_socket(
+ socket.create_connection((host, 443), timeout=10),
+ server_hostname=host,
+ )
+ key = base64.b64encode(os.urandom(16)).decode("ascii")
+ request_headers = {
+ "Host": host,
+ "Upgrade": "websocket",
+ "Connection": "Upgrade",
+ "Sec-WebSocket-Key": key,
+ "Sec-WebSocket-Version": "13",
+ "Origin": "https://www.formula1.com",
+ "User-Agent": "Mozilla/5.0",
+ }
+ request_headers.update(headers or {})
+ lines = [f"GET {path} HTTP/1.1", *[f"{k}: {v}" for k, v in request_headers.items()], "", ""]
+ self.sock.sendall("\r\n".join(lines).encode("utf-8"))
+ response = self._read_http_response()
+ if b" 101 " not in response.split(b"\r\n", 1)[0]:
+ raise RuntimeError(response.decode("utf-8", "replace"))
+
+ def _read_http_response(self) -> bytes:
+ data = b""
+ while b"\r\n\r\n" not in data:
+ data += self.sock.recv(4096)
+ return data
+
+ def send_text(self, text: str) -> None:
+ payload = text.encode("utf-8")
+ header = bytearray([0x81])
+ if len(payload) < 126:
+ header.append(0x80 | len(payload))
+ elif len(payload) < 65536:
+ header.append(0x80 | 126)
+ header.extend(struct.pack("!H", len(payload)))
+ else:
+ header.append(0x80 | 127)
+ header.extend(struct.pack("!Q", len(payload)))
+ mask = random.randbytes(4) if hasattr(random, "randbytes") else os.urandom(4)
+ masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
+ self.sock.sendall(bytes(header) + mask + masked)
+
+ def recv_text(self, timeout: float = 5.0) -> str | None:
+ self.sock.settimeout(timeout)
+ while True:
+ first = self.sock.recv(2)
+ if not first:
+ return None
+ opcode = first[0] & 0x0F
+ masked = bool(first[1] & 0x80)
+ length = first[1] & 0x7F
+ if length == 126:
+ length = struct.unpack("!H", self._read_exact(2))[0]
+ elif length == 127:
+ length = struct.unpack("!Q", self._read_exact(8))[0]
+ mask = self._read_exact(4) if masked else b""
+ payload = self._read_exact(length)
+ if masked:
+ payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
+ if opcode == 0x8:
+ return None
+ if opcode == 0x9:
+ self._send_pong(payload)
+ continue
+ if opcode == 0x1:
+ return payload.decode("utf-8", "replace")
+
+ def _send_pong(self, payload: bytes) -> None:
+ self.sock.sendall(bytes([0x8A, len(payload)]) + payload)
+
+ def _read_exact(self, n: int) -> bytes:
+ chunks = []
+ remaining = n
+ while remaining:
+ chunk = self.sock.recv(remaining)
+ if not chunk:
+ raise EOFError("socket closed")
+ chunks.append(chunk)
+ remaining -= len(chunk)
+ return b"".join(chunks)
+
+ def close(self) -> None:
+ self.sock.close()
+
+
+def http_json(url: str, method: str = "GET", headers: dict[str, str] | None = None) -> tuple[dict, list[str]]:
+ req = urllib.request.Request(url, method=method, headers=headers or {})
+ with urllib.request.urlopen(req, timeout=15) as resp:
+ cookies = resp.headers.get_all("Set-Cookie") or []
+ return json.loads(resp.read()), cookies
+
+
+def core_probe(seconds: int) -> None:
+ print("== SignalR Core ==")
+ neg, cookies = http_json(
+ CORE_NEGOTIATE,
+ method="POST",
+ headers={"Origin": "https://www.formula1.com", "User-Agent": "Mozilla/5.0", "Content-Length": "0"},
+ )
+ token = urllib.parse.quote(neg["connectionToken"], safe="")
+ ws = WebSocket(
+ CORE_WS_HOST,
+ f"/signalrcore?id={token}",
+ {"Cookie": "; ".join(cookies)} if cookies else None,
+ )
+ try:
+ ws.send_text(json.dumps({"protocol": "json", "version": 1}, separators=(",", ":")) + RS.decode())
+ print("handshake:", repr(ws.recv_text()))
+ ws.send_text(
+ json.dumps(
+ {"type": 1, "target": "subscribe", "arguments": [TOPICS], "invocationId": "1"},
+ separators=(",", ":"),
+ )
+ + RS.decode()
+ )
+ collect(ws, seconds)
+ finally:
+ ws.close()
+
+
+def legacy_probe(seconds: int, bearer: str | None) -> None:
+ print("\n== Legacy SignalR ==")
+ query = urllib.parse.quote(LEGACY_HUB, safe="")
+ headers = {"User-Agent": "BestHTTP"}
+ if bearer:
+ headers["Authorization"] = f"Bearer {bearer}"
+ neg_url = f"https://livetiming.formula1.com/signalr/negotiate?clientProtocol=1.5&connectionData={query}"
+ try:
+ neg, cookies = http_json(neg_url, headers=headers)
+ except Exception as exc:
+ print("negotiate failed:", exc)
+ return
+ token = urllib.parse.quote(neg["ConnectionToken"], safe="")
+ path = f"/signalr/connect?clientProtocol=1.5&transport=webSockets&connectionToken={token}&connectionData={query}"
+ ws_headers = {"User-Agent": "BestHTTP"}
+ if cookies:
+ ws_headers["Cookie"] = "; ".join(cookies)
+ if bearer:
+ ws_headers["Authorization"] = f"Bearer {bearer}"
+ ws = WebSocket(CORE_WS_HOST, path, ws_headers)
+ try:
+ ws.send_text(json.dumps({"H": "Streaming", "M": "Subscribe", "A": [TOPICS], "I": 1}, separators=(",", ":")))
+ collect(ws, seconds)
+ finally:
+ ws.close()
+
+
+def collect(ws: WebSocket, seconds: int) -> None:
+ deadline = time.time() + seconds
+ seen: dict[str, int] = {}
+ while time.time() < deadline:
+ try:
+ message = ws.recv_text(timeout=min(5, max(1, deadline - time.time())))
+ except TimeoutError:
+ continue
+ if not message:
+ break
+ for frame in message.split(RS.decode()):
+ if not frame:
+ continue
+ try:
+ payload = json.loads(frame)
+ except json.JSONDecodeError:
+ continue
+ record_topics(payload, seen)
+ print("seen topics:")
+ for topic in sorted(seen):
+ print(f" {topic}: {seen[topic]}")
+ if not any(topic.startswith("Position") for topic in seen):
+ print(" (no Position / Position.z topics observed)")
+
+
+def record_topics(payload: dict, seen: dict[str, int]) -> None:
+ result = payload.get("result") or payload.get("R")
+ if isinstance(result, dict):
+ for topic, data in result.items():
+ seen[topic] = seen.get(topic, 0) + 1
+ maybe_print_position(topic, data)
+ for message in payload.get("M") or []:
+ args = message.get("A") or []
+ if len(args) >= 2:
+ topic = args[0]
+ seen[topic] = seen.get(topic, 0) + 1
+ maybe_print_position(topic, args[1])
+ if payload.get("target") == "feed":
+ args = payload.get("arguments") or []
+ if len(args) >= 2:
+ topic = args[0]
+ seen[topic] = seen.get(topic, 0) + 1
+ maybe_print_position(topic, args[1])
+
+
+def maybe_print_position(topic: str, data) -> None:
+ if topic in {"Position", "Position.z"}:
+ text = json.dumps(data)[:600] if not isinstance(data, str) else data[:600]
+ print(f"POSITION TOPIC {topic}: {text}")
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--seconds", type=int, default=20)
+ parser.add_argument("--legacy", action="store_true", help="also probe legacy SignalR")
+ args = parser.parse_args()
+ core_probe(args.seconds)
+ if args.legacy:
+ legacy_probe(args.seconds, os.environ.get("BOXBOX_F1_LIVE_BEARER_TOKEN") or os.environ.get("F1_LIVE_BEARER_TOKEN"))
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())