""" TheTenths Telemetry Client ========================== Streams live lap and telemetry data to the TheTenths API. Supported games: * Automobilista 2 — pCars2 shared memory ($pcars2$ / $pcars$) * Assetto Corsa EVO (and AC) — Kunos shared memory (Local\\acpmf_*) Usage: python client.py [--game auto|ams2|acevo] [--token TOKEN] [--url URL] [--hz 10] Enable shared memory in the game: AMS2: Options → System → Shared Memory → "Project CARS 2" AC EVO: Settings → General → Shared Memory → On AC: Options → General → Shared Memory → On Requirements: pip install requests """ import argparse import ctypes import json import mmap import os import sys import time import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Optional import requests from ams2_shm import ( SharedMemory, SharedMemoryV1, SHARED_MEMORY_NAME, SHARED_MEMORY_NAME_V1, GAME_EXITED, PCARS2_MIN_VERSION, PCARS1_MIN_VERSION, PCARS1_MAX_VERSION, ) from ac_shm import ( ACPhysics, ACGraphics, ACStatic, AC_SHM_PHYSICS, AC_SHM_GRAPHICS, AC_SHM_STATIC, AC_OFF, AC_LIVE, AC_PAUSE, AC_SESSION_NAMES, ) # ── Helpers ─────────────────────────────────────────────────────────────────── _LAYOUT_ALIASES = {"GP": "Grand Prix", "Nat": "National", "Int": "International"} def _fmt_name(s: str) -> str: """CamelCase + underscore → readable words; expand common layout abbreviations.""" import re spaced = re.sub(r'([a-z])([A-Z])', r'\1 \2', s).replace('_', ' ') return ' '.join(_LAYOUT_ALIASES.get(w, w) for w in spaced.split()) def _fmt_track(base_raw: str, var_raw: str) -> str: base = _fmt_name(base_raw) if not var_raw or var_raw == base_raw: return base var = _fmt_name(var_raw) if var.lower().startswith(base.lower()): suffix = var[len(base):].strip() return f"{base} - {suffix}" if suffix else base return f"{base} - {var}" def _str(raw: bytes) -> str: return raw.rstrip(b"\x00").decode("utf-8", errors="replace") def _sec_to_ms(seconds: float) -> int: return int(seconds * 1000) def _temp_c(v: float) -> float: """AMS2 reports tyre/brake temps in Kelvin; convert to Celsius.""" return round(v - 273.15, 1) if v > 200 else round(v, 1) def _format_lap(ms: int) -> str: m = ms // 60_000 s = (ms % 60_000) // 1000 ms3 = ms % 1000 return f"{m}:{s:02d}.{ms3:03d}" # ── AMS2 shared memory reader ───────────────────────────────────────────────── class AMS2SharedMemory: """Opens AMS2 shared memory, auto-detecting pCars2 or pCars1 format.""" _CANDIDATES = [ (SHARED_MEMORY_NAME, SharedMemory, "pcars2", PCARS2_MIN_VERSION, 99), (SHARED_MEMORY_NAME_V1, SharedMemoryV1, "pcars1", PCARS1_MIN_VERSION, PCARS1_MAX_VERSION), ] def __init__(self): self._map: Optional[mmap.mmap] = None self._cls = None self._size = 0 self.fmt = None self.version = 0 self.tagname = None def open(self) -> bool: for name, cls, fmt, vmin, vmax in self._CANDIDATES: size = ctypes.sizeof(cls) try: m = mmap.mmap(0, size, tagname=name, access=mmap.ACCESS_READ) except Exception: continue try: m.seek(0) ver = int.from_bytes(m.read(4), "little") except Exception: m.close() continue if vmin <= ver <= vmax: self._map, self._cls, self._size = m, cls, size self.fmt, self.version, self.tagname = fmt, ver, name return True m.close() return False def read(self): if self._map is None: return None try: self._map.seek(0) raw = self._map.read(self._size) if len(raw) < self._size: raw = raw.ljust(self._size, b"\x00") return self._cls.from_buffer_copy(raw) except Exception as exc: print(f"[shm] AMS2 read error: {exc}") return None def close(self): if self._map: self._map.close() self._map = None self._cls = None self._size = 0 self.fmt = self.version = self.tagname = None # ── AC EVO shared memory reader ─────────────────────────────────────────────── class ACEvoSharedMemory: """Opens Assetto Corsa / AC EVO shared memory (all 3 sections).""" def __init__(self): self._phys_map: Optional[mmap.mmap] = None self._gfx_map: Optional[mmap.mmap] = None self._static_map: Optional[mmap.mmap] = None self._phys_size = ctypes.sizeof(ACPhysics) self._gfx_size = ctypes.sizeof(ACGraphics) self._static_size = ctypes.sizeof(ACStatic) def open(self) -> bool: try: self._gfx_map = mmap.mmap(0, self._gfx_size, tagname=AC_SHM_GRAPHICS, access=mmap.ACCESS_READ) self._phys_map = mmap.mmap(0, self._phys_size, tagname=AC_SHM_PHYSICS, access=mmap.ACCESS_READ) self._static_map = mmap.mmap(0, self._static_size, tagname=AC_SHM_STATIC, access=mmap.ACCESS_READ) return True except Exception: self.close() return False def read_graphics(self) -> Optional[ACGraphics]: return self._read(self._gfx_map, self._gfx_size, ACGraphics) def read_physics(self) -> Optional[ACPhysics]: return self._read(self._phys_map, self._phys_size, ACPhysics) def read_static(self) -> Optional[ACStatic]: return self._read(self._static_map, self._static_size, ACStatic) def _read(self, m, size, cls): if m is None: return None try: m.seek(0) raw = m.read(size) if len(raw) < size: raw = raw.ljust(size, b"\x00") return cls.from_buffer_copy(raw) except Exception as exc: print(f"[shm] AC EVO read error: {exc}") return None def close(self): for m in (self._phys_map, self._gfx_map, self._static_map): if m: try: m.close() except Exception: pass self._phys_map = self._gfx_map = self._static_map = None # ── API client ──────────────────────────────────────────────────────────────── class InvalidTokenError(Exception): """Raised when the API rejects the token (HTTP 401/403).""" class TelemetryAPI: def __init__(self, base_url: str, api_token: str, timeout: float = 3.0): self._base = base_url.rstrip("/") self._timeout = timeout self._session = requests.Session() self._session.headers.update({ "Content-Type": "application/json", "X-Api-Token": api_token, }) def ping(self) -> Optional[str]: """Validate token. Returns the account email on success, raises InvalidTokenError on 401/403.""" try: r = self._session.get(f"{self._base}/api/telemetry/ping", timeout=self._timeout) if r.status_code in (401, 403): raise InvalidTokenError(r.status_code) r.raise_for_status() return r.json().get("email", "unknown") except InvalidTokenError: raise except requests.RequestException as exc: print(f"[api] Token validation failed (network): {exc}") return None def session_start(self, payload: dict) -> bool: return self._post("/api/telemetry/session/start", payload) def live(self, payload: dict) -> bool: return self._post("/api/telemetry/live", payload) def lap_completed(self, payload: dict) -> bool: return self._post("/api/telemetry/lap", payload) def _post(self, path: str, payload: dict) -> bool: try: r = self._session.post( f"{self._base}{path}", json=payload, timeout=self._timeout, ) if r.status_code in (401, 403): raise InvalidTokenError(r.status_code) r.raise_for_status() return True except InvalidTokenError: raise except requests.RequestException as exc: print(f"[api] POST {path} failed: {exc}") return False # ── Session state ───────────────────────────────────────────────────────────── @dataclass class AMS2SessionState: session_id: str = field(default_factory=lambda: str(uuid.uuid4())) car: str = "" track: str = "" started: bool = False last_lap_number: int = -1 prev_s1: float = -1.0 prev_s2: float = -1.0 prev_s3: float = -1.0 @dataclass class ACSessionState: session_id: str = field(default_factory=lambda: str(uuid.uuid4())) car: str = "" track: str = "" started: bool = False last_laps: int = -1 prev_sector: int = -1 s1_ms: Optional[int] = None s2_ms: Optional[int] = None # ── Game-state name maps ────────────────────────────────────────────────────── _AMS2_STATE_NAMES = { 0: "EXITED", 1: "FRONT_END", 2: "INGAME_PLAYING", 3: "INGAME_PAUSED", 4: "INGAME_IN_MENU_TIME_TICKING", 5: "INGAME_RESTARTING", 6: "INGAME_REPLAY", 7: "FRONT_END_REPLAY", } _AC_STATUS_NAMES = {0: "OFF", 1: "REPLAY", 2: "LIVE", 3: "PAUSE"} # ── Game auto-detection ─────────────────────────────────────────────────────── def _detect_active_game(game_pref: str): """ Returns (game_name, shm_object) for whichever game has an active session, or None if neither is active. """ games = ["ams2", "acevo"] if game_pref == "auto" else [game_pref] for g in games: if g == "ams2": shm = AMS2SharedMemory() if shm.open(): data = shm.read() if data and int(data.mGameState) > GAME_EXITED and _str(data.mCarName): return ("ams2", shm) shm.close() elif g == "acevo": shm = ACEvoSharedMemory() if shm.open(): gfx = shm.read_graphics() if gfx and gfx.status == AC_LIVE: return ("acevo", shm) shm.close() return None # ── Diagnose ────────────────────────────────────────────────────────────────── def _diagnose(): print("\n[diagnose] AMS2 shared-memory sections:\n") _ams2_candidates = [ (SHARED_MEMORY_NAME, SharedMemory), (SHARED_MEMORY_NAME_V1, SharedMemoryV1), ] for name, cls in _ams2_candidates: size = ctypes.sizeof(cls) try: m = mmap.mmap(0, size, tagname=name, access=mmap.ACCESS_READ) except Exception as exc: print(f" {name:10s} -> open failed: {exc}") continue m.seek(0) raw = m.read(size).ljust(size, b"\x00") m.close() d = cls.from_buffer_copy(raw) gs = int(d.mGameState) print(f" {name:10s} ({cls.__name__:14s}, {size} B):") print(f" version={d.mVersion} build={d.mBuildVersionNumber} " f"gameState={gs}({_AMS2_STATE_NAMES.get(gs, '?')}) " f"sessionState={d.mSessionState}") if d.mVersion: print(f" car={_str(d.mCarName)!r} track={_str(d.mTrackLocation)!r} " f"speed={d.mSpeed*3.6:.1f} km/h gear={d.mGear} parts={d.mNumParticipants}") print() print("[diagnose] AC / AC EVO shared-memory sections:\n") _ac_sections = [ (AC_SHM_GRAPHICS, ACGraphics, "gfx"), (AC_SHM_PHYSICS, ACPhysics, "phys"), (AC_SHM_STATIC, ACStatic, "static"), ] for name, cls, label in _ac_sections: size = ctypes.sizeof(cls) try: m = mmap.mmap(0, size, tagname=name, access=mmap.ACCESS_READ) except Exception as exc: print(f" {name} -> open failed: {exc}") continue m.seek(0) raw = m.read(size).ljust(size, b"\x00") m.close() d = cls.from_buffer_copy(raw) print(f" {name} ({cls.__name__}, {size} B):") if label == "gfx": st = int(d.status) print(f" status={st}({_AC_STATUS_NAMES.get(st, '?')}) " f"session={d.session}({AC_SESSION_NAMES.get(d.session, '?')}) " f"laps={d.completedLaps} " f"iLastTime={d.iLastTime} ms compound={d.tyreCompound!r}") elif label == "phys": print(f" speed={d.speedKmh:.1f} km/h gear={d.gear} rpms={d.rpms} " f"fuel={d.fuel:.2f} brake={d.brake:.2f} gas={d.gas:.2f}") elif label == "static": print(f" car={d.carModel!r} track={d.track!r} " f"acVersion={d.acVersion!r} maxFuel={d.maxFuel:.1f}") print() # ── Main entry point ────────────────────────────────────────────────────────── def run(base_url: str, api_token: str, hz: int, debug: bool = False, diagnose: bool = False, game: str = "auto"): if diagnose: _diagnose() return interval = 1.0 / hz api = TelemetryAPI(base_url, api_token) print(f"[client] Validating API token ...") try: email = api.ping() if email is not None: print(f"[client] Token OK — logged in as {email}") else: print("[client] Could not reach server — will retry when a session is detected.") except InvalidTokenError: print() print("[api] API token rejected (401 Unauthorized).") print(" 1. Sign in at https://www.thetenths.com") print(" 2. Go to Profile -> Telemetry API Token -> Generate Token") print(" 3. Run the client again and paste the new token (or use --token).") return "invalid_token" game_label = {"ams2": "AMS2", "acevo": "AC EVO", "auto": "AMS2 or AC EVO"}[game] print(f"[client] Listening for {game_label} ...") while True: detected = _detect_active_game(game) if detected is None: print(f"[client] Waiting for an active {game_label} session ... (retry in 5 s)") time.sleep(5) continue game_name, shm = detected label = "AMS2" if game_name == "ams2" else "AC EVO" print(f"[client] {label} session detected — connecting ...") print(f" Sending to {base_url}") print(" Press Ctrl+C to quit.\n") try: if game_name == "ams2": _ams2_main_loop(shm, api, AMS2SessionState(), interval, debug) else: _ac_main_loop(shm, api, ACSessionState(), interval, debug) except KeyboardInterrupt: print("\n[client] Stopped by user.") return except InvalidTokenError: print() print("[api] Your API token was rejected (401 Unauthorized).") print(" Get a fresh token:") print(" 1. Sign in at https://www.thetenths.com") print(" 2. Go to Profile -> Telemetry API Token -> Generate Token") print(" 3. Run the client again and paste the new token " "(or use --token).") return "invalid_token" except Exception as exc: print(f"[client] Unexpected error: {exc} — reconnecting in 3 s") time.sleep(3) finally: shm.close() # ── AMS2 main loop ──────────────────────────────────────────────────────────── def _ams2_main_loop(shm: AMS2SharedMemory, api: TelemetryAPI, state: AMS2SessionState, interval: float, debug: bool = False): last_status_print = 0.0 while True: t0 = time.monotonic() data = shm.read() if data is None: break if data.mVersion == 0: print("[ams2] Shared memory went empty — reconnecting ...") break game_state = int(data.mGameState) car_name = _str(data.mCarName) game_active = game_state > GAME_EXITED and bool(car_name) if debug: gs_name = _AMS2_STATE_NAMES.get(game_state, str(game_state)) print(f"[dbg-ams2] ver={data.mVersion} state={game_state}({gs_name}) " f"car={car_name!r} track={_str(data.mTrackLocation)!r} " f"speed={data.mSpeed*3.6:.1f} km/h gear={data.mGear}") if not game_active: if state.started: print("[ams2] Left session — back in menu.") state = AMS2SessionState() now = time.monotonic() if now - last_status_print >= 5.0: gs_name = _AMS2_STATE_NAMES.get(game_state, str(game_state)) print(f"[ams2] Waiting for session ... (state={game_state}={gs_name} car={car_name!r})") last_status_print = now time.sleep(1) continue car = car_name track = _fmt_track(_str(data.mTrackLocation), _str(data.mTrackVariation)) if not state.started or car != state.car or track != state.track: state.session_id = str(uuid.uuid4()) state.car = car state.track = track state.started = True state.last_lap_number = -1 api.session_start({ "session_id": state.session_id, "car": car, "track": track, "game": "AMS2", "session_type": _ams2_session_type(data.mSessionState), }) print(f"[ams2] Session started — {car} @ {track} (id={state.session_id[:8]})") idx = int(data.mViewedParticipantIndex) n = int(data.mNumParticipants) pidx = max(0, min(idx, n - 1)) if n > 0 else 0 part = data.mParticipantInfo[pidx] current_lap = int(part.mCurrentLap) s1_now = data.mCurrentSector1Time s2_now = data.mCurrentSector2Time s3_now = data.mCurrentSector3Time if state.last_lap_number >= 0 and current_lap > state.last_lap_number: laps_done = int(part.mLapsCompleted) last_ms = _sec_to_ms(data.mLastLapTime) if data.mLastLapTime > 0 else None is_valid = (not bool(data.mLapInvalidated)) and last_ms is not None compound = "" if hasattr(data, "mTyreCompound"): compound = _str(bytes(data.mTyreCompound[0])) def _s(v: float): return _sec_to_ms(v) if v > 0 else None api.lap_completed({ "session_id": state.session_id, "lap_number": laps_done, "lap_time_ms": last_ms, "is_valid": is_valid, "player": _str(part.mName) or "Player", "car": car, "track": track, "game": "AMS2", "sim_version": str(int(data.mBuildVersionNumber)), "tyre_compound": compound, "sector1_ms": _s(state.prev_s1), "sector2_ms": _s(state.prev_s2), "sector3_ms": _s(state.prev_s3), "tyre_temp_f": [_temp_c(data.mTyreTreadTemp[i]) for i in range(4)], "brake_temp_f": [_temp_c(data.mBrakeTempCelsius[i]) for i in range(4)], "fuel_level": round(data.mFuelLevel, 2), "fuel_capacity": round(data.mFuelCapacity, 2), }) marker = "OK " if is_valid else "INV" lap_str = _format_lap(last_ms) if last_ms else "n/a" comp_str = f" ({compound})" if compound else "" print(f" [{marker}] Lap {laps_done:>3} — {lap_str}{comp_str}") state.prev_s1 = s1_now state.prev_s2 = s2_now state.prev_s3 = s3_now state.last_lap_number = current_lap speed_kmh = data.mSpeed * 3.6 current_ms = _sec_to_ms(data.mCurrentTime) if data.mCurrentTime >= 0 else None best_ms = _sec_to_ms(data.mBestLapTime) if data.mBestLapTime > 0 else None susp = [round(data.mSuspensionTravel[i], 3) for i in range(4)] \ if hasattr(data, "mSuspensionTravel") else None api.live({ "session_id": state.session_id, "speed_kmh": round(speed_kmh, 1), "gear": data.mGear, "rpm": round(data.mRpm), "max_rpm": round(data.mMaxRPM), "throttle": round(data.mThrottle, 3), "brake": round(data.mBrake, 3), "clutch": round(data.mClutch, 3), "lap_time_ms": current_ms, "best_lap_ms": best_ms, "current_lap": current_lap, "fuel_level": round(data.mFuelLevel, 2), "oil_temp": round(data.mOilTempCelsius, 1), "water_temp": round(data.mWaterTempCelsius, 1), "tyre_temp": [_temp_c(data.mTyreTreadTemp[i]) for i in range(4)], "suspension_travel": susp, "ambient_temp": round(data.mAmbientTemperature, 1), "track_temp": round(data.mTrackTemperature, 1), }) elapsed = time.monotonic() - t0 sleep = interval - elapsed if sleep > 0: time.sleep(sleep) # ── AC EVO main loop ────────────────────────────────────────────────────────── def _ac_main_loop(shm: ACEvoSharedMemory, api: TelemetryAPI, state: ACSessionState, interval: float, debug: bool = False): last_status_print = 0.0 while True: t0 = time.monotonic() gfx = shm.read_graphics() phys = shm.read_physics() static = shm.read_static() if gfx is None or phys is None or static is None: break if gfx.status not in (AC_LIVE, AC_PAUSE): if state.started: print("[acevo] Left session.") state = ACSessionState() now = time.monotonic() if now - last_status_print >= 5.0: st_name = _AC_STATUS_NAMES.get(gfx.status, str(gfx.status)) print(f"[acevo] Waiting for session ... (status={gfx.status}={st_name})") last_status_print = now time.sleep(1) continue car = static.carModel.strip() track = _fmt_track(static.track.strip(), static.trackConfiguration.strip()) if not state.started or car != state.car or track != state.track: state.session_id = str(uuid.uuid4()) state.car = car state.track = track state.started = True state.last_laps = gfx.completedLaps state.prev_sector = gfx.currentSectorIndex state.s1_ms = None state.s2_ms = None api.session_start({ "session_id": state.session_id, "car": car, "track": track, "game": "ACEVO", "session_type": AC_SESSION_NAMES.get(gfx.session, "unknown"), }) print(f"[acevo] Session started — {car} @ {track} (id={state.session_id[:8]})") if debug: st_name = _AC_STATUS_NAMES.get(gfx.status, str(gfx.status)) print(f"[dbg-acevo] status={gfx.status}({st_name}) laps={gfx.completedLaps} " f"sector={gfx.currentSectorIndex} lastSector={gfx.lastSectorTime} ms " f"speed={phys.speedKmh:.1f} km/h gear={phys.gear} rpms={phys.rpms}") # ── Sector time capture on index transition ────────────────────── cur_sector = gfx.currentSectorIndex if state.prev_sector >= 0 and cur_sector != state.prev_sector: st = gfx.lastSectorTime if gfx.lastSectorTime > 0 else None if state.prev_sector == 0: state.s1_ms = st elif state.prev_sector == 1: state.s2_ms = st state.prev_sector = cur_sector # ── Lap completed ──────────────────────────────────────────────── if state.last_laps >= 0 and gfx.completedLaps > state.last_laps: last_ms = gfx.iLastTime if gfx.iLastTime > 0 else None is_valid = last_ms is not None s1 = state.s1_ms s2 = state.s2_ms # Derive s3 from total; fall back to lastSectorTime at crossing if last_ms and s1 and s2: s3: Optional[int] = last_ms - s1 - s2 elif gfx.lastSectorTime > 0: s3 = gfx.lastSectorTime else: s3 = None compound = gfx.tyreCompound.strip() player = f"{static.playerName} {static.playerSurname}".strip() or "Player" api.lap_completed({ "session_id": state.session_id, "lap_number": gfx.completedLaps, "lap_time_ms": last_ms, "is_valid": is_valid, "player": player, "car": car, "track": track, "game": "ACEVO", "sim_version": static.acVersion.strip(), "tyre_compound": compound, "sector1_ms": s1, "sector2_ms": s2, "sector3_ms": s3, "tyre_temp_f": [round(phys.tyreCoreTemperature[i], 1) for i in range(4)], "brake_temp_f": [round(phys.brakeTemp[i], 1) for i in range(4)], "fuel_level": round(phys.fuel, 2), "fuel_capacity": round(static.maxFuel, 2), }) marker = "OK " if is_valid else "INV" lap_str = _format_lap(last_ms) if last_ms else "n/a" comp_str = f" ({compound})" if compound else "" print(f" [{marker}] Lap {gfx.completedLaps:>3} — {lap_str}{comp_str}") state.s1_ms = None state.s2_ms = None state.last_laps = gfx.completedLaps # ── Live telemetry ─────────────────────────────────────────────── current_ms = gfx.iCurrentTime if gfx.iCurrentTime > 0 else None best_ms = gfx.iBestTime if gfx.iBestTime > 0 else None api.live({ "session_id": state.session_id, "speed_kmh": round(phys.speedKmh, 1), "gear": phys.gear, "rpm": phys.rpms, "max_rpm": static.maxRpm, "throttle": round(phys.gas, 3), "brake": round(phys.brake, 3), "clutch": round(phys.clutch, 3), "lap_time_ms": current_ms, "best_lap_ms": best_ms, "current_lap": gfx.completedLaps, "fuel_level": round(phys.fuel, 2), "oil_temp": None, "water_temp": None, "tyre_temp": [round(phys.tyreCoreTemperature[i], 1) for i in range(4)], "suspension_travel": [round(phys.suspensionTravel[i], 3) for i in range(4)], "ambient_temp": round(phys.airTemp, 1), "track_temp": round(phys.roadTemp, 1), }) elapsed = time.monotonic() - t0 sleep = interval - elapsed if sleep > 0: time.sleep(sleep) # ── Session type helpers ────────────────────────────────────────────────────── def _ams2_session_type(session_state: int) -> str: return {0: "invalid", 1: "practice", 2: "test", 3: "qualify", 4: "formation_lap", 5: "race", 6: "time_attack" }.get(int(session_state), "unknown") # ── Entry point helpers ─────────────────────────────────────────────────────── def _is_frozen() -> bool: """True when running as a packaged .exe (PyInstaller or Nuitka).""" return getattr(sys, "frozen", False) or "__compiled__" in dir() def _config_path() -> Path: base = os.environ.get("APPDATA") or os.path.expanduser("~") return Path(base) / "TheTenths" / "telemetry.json" def _load_saved_token() -> Optional[str]: try: p = _config_path() if p.is_file(): tok = (json.loads(p.read_text(encoding="utf-8")).get("token") or "").strip() return tok or None except Exception: pass return None def _save_token(token: str) -> Optional[Path]: try: p = _config_path() p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps({"token": token}), encoding="utf-8") return p except Exception as exc: print(f"[config] Could not save token: {exc}") return None def _forget_token(): try: p = _config_path() if p.is_file(): p.unlink() print(f"[config] Removed saved token at {p}") else: print("[config] No saved token to remove.") except Exception as exc: print(f"[config] Could not remove token: {exc}") def _pause_if_frozen(): if _is_frozen(): try: input("\nClient stopped. Press Enter to close ...") except EOFError: pass def main(): parser = argparse.ArgumentParser( description="Stream sim racing telemetry to TheTenths API" ) parser.add_argument("--game", default="auto", choices=["auto", "ams2", "acevo"], help="Game to read from: auto (default), ams2, acevo") parser.add_argument("--url", default="https://www.thetenths.com", help="Base URL of the TheTenths service " "(default: https://www.thetenths.com; use " "http://localhost:8080 for local dev)") parser.add_argument("--token", default=None, help="API token from your TheTenths profile page " "(prompted for if omitted)") parser.add_argument("--hz", type=int, default=10, help="Live telemetry send rate in Hz (default 10)") parser.add_argument("--debug", action="store_true", help="Print raw telemetry every frame") parser.add_argument("--diagnose", action="store_true", help="Scan all shared memory sections and print values, then exit") parser.add_argument("--forget-token", action="store_true", help="Delete the saved API token and exit") args = parser.parse_args() if sys.platform != "win32": print("Error: shared memory telemetry is only available on Windows.") sys.exit(1) if args.forget_token: _forget_token() _pause_if_frozen() return token = (args.token or "").strip() or None saved_used = False if not token and not args.diagnose: token = _load_saved_token() if token: saved_used = True print(f"[config] Using saved API token from {_config_path()}") print(" (run with --forget-token to clear it)") else: try: token = input("Enter your TheTenths API token " "(Profile -> Telemetry API Token): ").strip() except EOFError: token = "" if not token: print("No token provided - exiting.") _pause_if_frozen() sys.exit(1) if token and not args.diagnose and not saved_used: saved = _save_token(token) if saved: print(f"[config] Saved API token to {saved}") try: result = run(args.url, token or "diagnose", args.hz, args.debug, args.diagnose, args.game) if result == "invalid_token": _forget_token() finally: _pause_if_frozen() if __name__ == "__main__": main()