""" SimOS iRacing Live Telemetry WebSocket Bridge & Native Python Audio Engine ------------------------------------------------------------------------- Connects to iRacing Shared Memory (via pyirsdk / irsdk) and broadcasts high-frequency 60Hz telemetry JSON packets to SimOS Cockpit at ws://localhost:8088. Native Hardware & Audio Architecture: - Direct Windows host soundcard enumeration via sounddevice.query_devices() - Low-latency native WASAPI / DirectSound audio playback (sounddevice.play) - Real-time continuous 60Hz tactile streaming engine with anti-mud priority ducking - Full 300+ iRacing telemetry channel ingestion dynamically exported """ import sys import os class UnbufferedStream: def __init__(self, file_path: str): self.f = open(file_path, "a", encoding="utf-8") def write(self, data: str): try: self.f.write(data) self.f.flush() except Exception: pass def flush(self): try: self.f.flush() except Exception: pass def __getattr__(self, name: str): return getattr(self.f, name) # Under pythonw.exe on Windows, sys.stdout and sys.stderr are None. # Redirect immediately so any print/log statement does not crash pythonw. if sys.stdout is None: try: log_dir = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS") if os.path.exists(log_dir): sys.stdout = UnbufferedStream(os.path.join(log_dir, "bridge_output.log")) else: sys.stdout = open(os.devnull, "w", encoding="utf-8") except Exception: try: sys.stdout = open(os.devnull, "w", encoding="utf-8") except Exception: pass if sys.stderr is None: try: if sys.stdout and not getattr(sys.stdout, "closed", False): sys.stderr = sys.stdout else: sys.stderr = open(os.devnull, "w", encoding="utf-8") except Exception: pass if sys.stdout is not None and hasattr(sys.stdout, "reconfigure"): try: sys.stdout.reconfigure(line_buffering=True) except Exception: pass if sys.stderr is not None and hasattr(sys.stderr, "reconfigure"): try: sys.stderr.reconfigure(line_buffering=True) except Exception: pass # Single-Instance Mutex (Windows Kernel Named Mutex) # Guarantees only ONE bridge instance runs at a time; duplicate launches exit immediately with 0 tray icons. is_sync_cli = any(arg in ("--sync", "--sync-rig") for arg in sys.argv) if sys.platform == "win32" and __name__ == "__main__" and not is_sync_cli: try: import ctypes MUTEX_NAME = "Local\\SimOS_Companion_SingleInstance_Mutex" kernel32 = ctypes.windll.kernel32 single_instance_mutex = kernel32.CreateMutexW(None, False, MUTEX_NAME) if kernel32.GetLastError() == 183: # ERROR_ALREADY_EXISTS try: print("[SimOS] Another instance is already running. Exiting.") except Exception: pass sys.exit(0) ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("GridPass.SimOS") except Exception: pass import time import json import asyncio import math import urllib.request import urllib.parse import urllib.error import webbrowser import threading import subprocess import hashlib import configparser import glob import re import struct import socket from typing import Set, Any, Dict, List, Optional, Tuple try: import win32gui import win32con HAS_WIN32 = True except Exception: HAS_WIN32 = False try: from winotify import Notification HAS_WINOTIFY = True except Exception: HAS_WINOTIFY = False try: import bleak from bleak import BleakScanner HAS_BLEAK = True except Exception: HAS_BLEAK = False BleakScanner = None def global_excepthook(exc_type, exc_value, exc_tb): if issubclass(exc_type, KeyboardInterrupt): sys.__excepthook__(exc_type, exc_value, exc_tb) return import traceback print(f"\n[UNHANDLED EXCEPTION] {exc_type.__name__}: {exc_value}") traceback.print_exception(exc_type, exc_value, exc_tb) sys.excepthook = global_excepthook if hasattr(threading, 'excepthook'): threading.excepthook = lambda args: global_excepthook(args.exc_type, args.exc_value, args.exc_traceback) BRIDGE_VERSION = "1.9.4" def parse_semver(v: str) -> tuple: """Parse version string into comparable tuple of ints.""" try: parts = [int(p) for p in v.strip().lstrip("v").split(".") if p.isdigit()] return tuple(parts) except Exception: return (0, 0, 0) def sanitize_for_json(val: Any) -> Any: """Recursively converts non-JSON-serializable types (bytes, memoryviews, NaN/inf) to safe primitives.""" if val is None: return None if isinstance(val, (bool, str, int)): return val if isinstance(val, float): if math.isnan(val) or math.isinf(val): return 0.0 return round(val, 5) if isinstance(val, (bytes, bytearray, memoryview)): try: return bytes(val).decode("utf-8", errors="replace").rstrip("\x00") except Exception: return str(val) if isinstance(val, (list, tuple)): return [sanitize_for_json(item) for item in val] if isinstance(val, dict): return {str(k): sanitize_for_json(v) for k, v in val.items()} try: return float(val) except Exception: try: return str(val) except Exception: return None # Ensure Windows stdout handles UTF-8 cleanly if sys.platform == "win32": try: if sys.stdout and hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") if sys.stderr and hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8", errors="replace") except Exception: pass # Import sounddevice & numpy for Native Windows Audio Engine HAS_SOUNDDEVICE = False try: import sounddevice as sd import numpy as np HAS_SOUNDDEVICE = True except ImportError: HAS_SOUNDDEVICE = False print("[WARN] sounddevice or numpy not installed. Native audio engine running in fallback mode.") # Import irsdk (or pyirsdk) HAS_IRSDK = False try: import irsdk HAS_IRSDK = True except ImportError: try: import pyirsdk as irsdk HAS_IRSDK = True except ImportError: HAS_IRSDK = False try: import websockets except ImportError: print("[ERROR] 'websockets' library is required. Install via: pip install websockets") sys.exit(1) # Import pyserial for Native ESP32 USB CDC Telemetry Streaming HAS_SERIAL = False try: import serial import serial.tools.list_ports HAS_SERIAL = True except ImportError: HAS_SERIAL = False print("[WARN] pyserial not installed. ESP32 hardware streaming running in fallback mode.") # Import esptool for 1-Click ESP32-S3 Firmware Flashing HAS_ESPTOOL = False try: import esptool HAS_ESPTOOL = True except ImportError: HAS_ESPTOOL = False # Import pystray & Pillow for Windows System Tray Icon & 1-Click Cockpit Launcher HAS_TRAY = False try: from PIL import Image, ImageDraw, ImageTk import pystray from pystray import MenuItem as item, Menu HAS_TRAY = True except ImportError: HAS_TRAY = False # Import Tkinter and qrcode for In-Game HUD & Scan-to-Race QR Code Overlay HAS_TKINTER = False try: import tkinter as tk HAS_TKINTER = True except Exception: HAS_TKINTER = False HAS_QRCODE = False try: import qrcode HAS_QRCODE = True except Exception: HAS_QRCODE = False # Connected WebSocket clients CLIENTS: Set[Any] = set() # ============================================================================ # 1. NATIVE WINDOWS SOUNDCARD ENUMERATION & PLAYBACK ENGINE # ============================================================================ AUDIO_DEVICES_CACHE: List[Dict[str, Any]] = [] LAST_DEVICE_QUERY = 0.0 def get_audio_output_devices(force_refresh: bool = False) -> List[Dict[str, Any]]: """Queries all host audio output devices via sounddevice, prioritizing Windows WASAPI.""" global AUDIO_DEVICES_CACHE, LAST_DEVICE_QUERY now = time.time() if not force_refresh and AUDIO_DEVICES_CACHE and (now - LAST_DEVICE_QUERY < 10.0): return AUDIO_DEVICES_CACHE if not HAS_SOUNDDEVICE: return [] devices_list = [] try: all_devs = sd.query_devices() hostapis = sd.query_hostapis() # Find WASAPI host API index wasapi_idx = next((i for i, h in enumerate(hostapis) if 'WASAPI' in h.get('name', '').upper()), None) # If no WASAPI, fallback to DirectSound or default host API target_hostapi = wasapi_idx if target_hostapi is None: target_hostapi = next((i for i, h in enumerate(hostapis) if 'DIRECTSOUND' in h.get('name', '').upper()), None) seen_names = set() for idx, d in enumerate(all_devs): max_out = d.get("max_output_channels", 0) if max_out < 1: continue # If target host API exists, filter to ONLY those matching it if target_hostapi is not None and d.get("hostapi") != target_hostapi: continue raw_name = str(d.get("name", "")).strip() # Exclude blank, placeholders like 'Output ()', or names starting with '@' if not raw_name or raw_name.startswith("@"): continue name_lower = raw_name.lower() if name_lower.startswith("output (") or name_lower == "output ()" or name_lower == "primary sound driver": continue if "bthhfenum.sys" in name_lower: continue # Deduplicate by clean name to guarantee 100% unique devices if name_lower in seen_names: continue seen_names.add(name_lower) h_name = "Audio" if isinstance(hostapis, (list, tuple)) and 0 <= d.get("hostapi", 0) < len(hostapis): h_name = hostapis[d.get("hostapi", 0)].get("name", "Audio") devices_list.append({ "index": idx, "name": raw_name, "hostapi_name": h_name, "max_output_channels": int(max_out), "default_samplerate": float(d.get("default_samplerate", 48000.0)), }) AUDIO_DEVICES_CACHE = devices_list LAST_DEVICE_QUERY = now except Exception as e: print(f"[AUDIO] Device query error: {e}") return AUDIO_DEVICES_CACHE def play_native_test_tone( device_index: Optional[Any] = None, freq: float = 45.0, gain: float = 0.8, duration: float = 1.0, stereo_channel: str = "MONO" ): """Generates and plays an ultra-clean envelope-shaped sine test tone directly to Windows WASAPI.""" if not HAS_SOUNDDEVICE: print("[AUDIO] sounddevice is not installed.") return try: target_dev = None if device_index is not None: raw_str = str(device_index).strip() if raw_str.startswith("native:"): raw_str = raw_str.replace("native:", "") if raw_str.isdigit(): target_dev = int(raw_str) sr = 48000 if target_dev is not None: try: d_info = sd.query_devices(target_dev) sr = int(d_info.get("default_samplerate", 48000)) except Exception: target_dev = None freq_val = max(15.0, min(180.0, float(freq))) gain_val = max(0.01, min(1.0, float(gain))) dur_val = max(0.1, min(5.0, float(duration))) t = np.linspace(0, dur_val, int(sr * dur_val), endpoint=False, dtype=np.float32) waveform = np.sin(2 * np.pi * freq_val * t) # Anti-click envelope: 30ms fade-in, 60ms fade-out fade_in = int(sr * 0.03) fade_out = int(sr * 0.06) envelope = np.ones_like(waveform) if fade_in > 0: envelope[:fade_in] = np.linspace(0.0, 1.0, fade_in) if fade_out > 0: envelope[-fade_out:] = np.linspace(1.0, 0.0, fade_out) scaled = waveform * envelope * gain_val stereo_upper = str(stereo_channel).upper() if stereo_upper == "L": out_buf = np.column_stack((scaled, np.zeros_like(scaled))) elif stereo_upper == "R": out_buf = np.column_stack((np.zeros_like(scaled), scaled)) else: out_buf = np.column_stack((scaled, scaled)) # Asynchronous non-blocking playback to host soundcard sd.play(out_buf, samplerate=sr, device=target_dev) except Exception as e: print(f"[NATIVE AUDIO ERROR] {e}") # ============================================================================ # 2. CONTINUOUS LIVE TACTILE STREAMING & ANTI-MUD DUCKING ENGINE # ============================================================================ class NativeTactileEngine: """Manages active sounddevice OutputStream instances and runs 60Hz telemetry audio synthesis.""" def __init__(self): self.channels: List[Dict[str, Any]] = [] self.master_gain: float = 0.90 self.muted: bool = False self.enabled: bool = True self.streams: Dict[Optional[int], Any] = {} self.targets: Dict[Optional[int], Dict[str, List[tuple]]] = {} self.phases: Dict[Optional[int], Dict[str, float]] = {} def update_config(self, channels: list, master_gain: float = 90.0, muted: bool = False, enabled: bool = True): self.channels = channels if isinstance(channels, list) else [] self.master_gain = max(0.0, min(1.0, float(master_gain) / 100.0)) self.muted = bool(muted) self.enabled = bool(enabled) def _get_dev_key(self, ch: dict) -> Optional[int]: raw_dev = ch.get("soundCardDeviceIndex") if raw_dev is None: raw_dev = ch.get("soundCardDeviceId") if raw_dev is not None: s = str(raw_dev).strip() if s.startswith("native:"): s = s.replace("native:", "") if s.isdigit(): return int(s) return None def ensure_stream(self, dev_key: Optional[int]): if not HAS_SOUNDDEVICE: return if dev_key in self.streams: stream = self.streams[dev_key] if stream and getattr(stream, 'active', False): return def audio_cb(outdata, frames, time_info, status): sr = 48000 outdata.fill(0) if not self.enabled or self.muted: return t_info = self.targets.get(dev_key, {"L": [], "R": []}) if dev_key not in self.phases: self.phases[dev_key] = {"L": 0.0, "R": 0.0} t = np.arange(frames, dtype=np.float32) / sr sig_l = np.zeros(frames, dtype=np.float32) sig_r = np.zeros(frames, dtype=np.float32) # Synthesize Left Channel for freq, g in t_info.get("L", []): if g > 0.005: sig_l += np.sin(2 * np.pi * freq * t + self.phases[dev_key]["L"]) * g if t_info.get("L"): self.phases[dev_key]["L"] = (self.phases[dev_key]["L"] + 2 * np.pi * t_info["L"][0][0] * frames / sr) % (2 * np.pi) # Synthesize Right Channel for freq, g in t_info.get("R", []): if g > 0.005: sig_r += np.sin(2 * np.pi * freq * t + self.phases[dev_key]["R"]) * g if t_info.get("R"): self.phases[dev_key]["R"] = (self.phases[dev_key]["R"] + 2 * np.pi * t_info["R"][0][0] * frames / sr) % (2 * np.pi) # Soft clip / limiter outdata[:, 0] = np.clip(sig_l * 0.75, -0.98, 0.98) outdata[:, 1] = np.clip(sig_r * 0.75, -0.98, 0.98) try: stream = sd.OutputStream( device=dev_key, channels=2, samplerate=48000, blocksize=512, callback=audio_cb ) stream.start() self.streams[dev_key] = stream except Exception: pass def evaluate_frame(self, telemetry_packet: dict): """Processes 60Hz telemetry frame and updates frequency/gain targets with Anti-Mud ducking.""" if not HAS_SOUNDDEVICE or not self.enabled or self.muted or not self.channels: return kerb_l = telemetry_packet.get("kerbRumbleLeft", 0) kerb_r = telemetry_packet.get("kerbRumbleRight", 0) rpm = telemetry_packet.get("rpm", 0) max_rpm = 12500 shift_active = rpm > (max_rpm * 0.94) abs_active = bool(telemetry_packet.get("absActive", False)) traction_active = bool(telemetry_packet.get("tractionLossActive", False)) speed_mph = telemetry_packet.get("speedMph", 0) new_targets: Dict[Optional[int], Dict[str, List[tuple]]] = {} for ch in self.channels: if not isinstance(ch, dict) or ch.get("muted", False): continue dev_key = self._get_dev_key(ch) if dev_key not in new_targets: new_targets[dev_key] = {"L": [], "R": []} stereo = str(ch.get("stereoChannel", "MONO")).upper() is_left = stereo in ["L", "MONO"] is_right = stereo in ["R", "MONO"] ch_gain = (float(ch.get("gain", 80)) / 100.0) * self.master_gain effects = ch.get("effects", []) # Check transient activity for anti-mud ducking has_p1 = False for eff in effects: if eff.get("enabled", True) and eff.get("priority") == "priority_1_transient": t_type = eff.get("type", "") if (t_type == "gear_shift" and shift_active) or (t_type == "kerb_strike" and max(kerb_l, kerb_r) > 15): has_p1 = True break duck_factor = 1.0 if has_p1 and ch.get("duckingEnabled", True): duck_factor = 1.0 - (float(ch.get("duckingAttenuationPct", 75)) / 100.0) for eff in effects: if not eff.get("enabled", True): continue eff_type = eff.get("type", "") active = False base_val = 0.0 if eff_type == "gear_shift": active = shift_active base_val = 1.0 elif eff_type == "kerb_strike": active = max(kerb_l, kerb_r) > 10 base_val = min(1.0, max(kerb_l, kerb_r) / 100.0) elif eff_type == "abs_chatter": active = abs_active base_val = 0.85 elif eff_type == "traction_loss": active = traction_active base_val = 0.85 elif eff_type == "engine_rpm": active = rpm > 600 base_val = min(1.0, rpm / max_rpm) elif eff_type == "road_texture": active = speed_mph > 5 base_val = min(1.0, speed_mph / 160.0) if active: priority = eff.get("priority", "priority_3_ambient") e_gain = (float(eff.get("gain", 80)) / 100.0) * base_val * ch_gain if priority != "priority_1_transient": e_gain *= duck_factor freq = float(eff.get("frequency", 45.0)) if is_left: new_targets[dev_key]["L"].append((freq, e_gain)) if is_right: new_targets[dev_key]["R"].append((freq, e_gain)) self.targets = new_targets for dev_key in new_targets.keys(): self.ensure_stream(dev_key) NATIVE_ENGINE = NativeTactileEngine() # ============================================================================ # 3. TELEMETRY EXTRACTION (300+ CHANNELS) & BROADCASTING # ============================================================================ def get_channel(ir_instance: Any, key: str, fallback: Any = 0.0) -> Any: """Safely reads a channel from irsdk shared memory dictionary.""" try: if ir_instance and hasattr(ir_instance, '_var_headers_dict') and ir_instance._var_headers_dict: if key in ir_instance._var_headers_dict: val = ir_instance[key] return val if val is not None else fallback except Exception: pass return fallback def extract_car_session_info(ir_instance: Any) -> Dict[str, Any]: """Reads car and track session info from iRacing YAML session data.""" car_path = "" car_screen_name = "" car_class_short_name = "" track_name = "" track_display_name = "" track_config_name = "" driver_name = "" car_number = "" first_rpm = 0.0 shift_rpm = 0.0 last_rpm = 0.0 blink_rpm = 0.0 redline = 0.0 idle_rpm = 0.0 def _parse_num(val: Any) -> float: try: if val is not None: f = float(val) if not math.isnan(f) and f > 0: return f except Exception: pass return 0.0 if not ir_instance: engine_limits = { "minRpm": 3500.0, "shiftRpm": 7000.0, "redlineRpm": 7500.0, "idleRpm": 800.0, "carName": "Active Vehicle", } return { "CarPath": car_path, "CarScreenName": car_screen_name, "carScreenName": car_screen_name, "CarClassShortName": car_class_short_name, "carClassShortName": car_class_short_name, "TrackName": track_name, "trackName": track_name, "TrackDisplayName": track_display_name, "trackDisplayName": track_display_name, "TrackConfigName": track_config_name, "trackConfigName": track_config_name, "DriverName": driver_name, "driverName": driver_name, "CarNumber": car_number, "carNumber": car_number, "DriverCarSLFirstRPM": first_rpm, "DriverCarSLShiftRPM": shift_rpm, "DriverCarSLLastRPM": last_rpm, "DriverCarSLBlinkRPM": blink_rpm, "DriverCarRedLine": redline, "DriverCarIdleRPM": idle_rpm, "engineLimits": engine_limits, "engine_limits": engine_limits, } try: weekend_info = ir_instance['WeekendInfo'] if isinstance(weekend_info, dict): track_name = str(weekend_info.get('TrackName', '') or '') track_display_name = str(weekend_info.get('TrackDisplayName', '') or weekend_info.get('TrackName', '') or '') track_config_name = str(weekend_info.get('TrackConfigName', '') or '') except Exception: pass try: driver_info = ir_instance['DriverInfo'] if isinstance(driver_info, dict): driver_idx = driver_info.get('DriverCarIdx', 0) drivers = driver_info.get('Drivers', []) player_driver = None if isinstance(drivers, list) and drivers: for d in drivers: if isinstance(d, dict) and d.get('CarIdx') == driver_idx: player_driver = d break if not player_driver and isinstance(driver_idx, int) and 0 <= driver_idx < len(drivers): if isinstance(drivers[driver_idx], dict): player_driver = drivers[driver_idx] if not player_driver and isinstance(drivers[0], dict): player_driver = drivers[0] if player_driver: car_path = str(player_driver.get('CarPath', '') or '') car_screen_name = str(player_driver.get('CarScreenName', '') or player_driver.get('CarScreenNameShort', '') or '') car_class_short_name = str(player_driver.get('CarClassShortName', '') or '') driver_name = str(player_driver.get('UserName', '') or '') car_number = str(player_driver.get('CarNumber', '') or '') # Extract Shift Light & Engine Limit RPMs from DriverInfo (with player_driver fallback) first_rpm = _parse_num(driver_info.get('DriverCarSLFirstRPM')) or (player_driver and _parse_num(player_driver.get('DriverCarSLFirstRPM'))) or 0.0 shift_rpm = _parse_num(driver_info.get('DriverCarSLShiftRPM')) or (player_driver and _parse_num(player_driver.get('DriverCarSLShiftRPM'))) or 0.0 last_rpm = _parse_num(driver_info.get('DriverCarSLLastRPM')) or (player_driver and _parse_num(player_driver.get('DriverCarSLLastRPM'))) or 0.0 blink_rpm = _parse_num(driver_info.get('DriverCarSLBlinkRPM')) or (player_driver and _parse_num(player_driver.get('DriverCarSLBlinkRPM'))) or 0.0 redline = _parse_num(driver_info.get('DriverCarRedLine')) or (player_driver and _parse_num(player_driver.get('DriverCarRedLine'))) or 0.0 idle_rpm = _parse_num(driver_info.get('DriverCarIdleRPM')) or (player_driver and _parse_num(player_driver.get('DriverCarIdleRPM'))) or 0.0 except Exception: pass engine_limits = { "minRpm": float(first_rpm or (redline * 0.55 if redline else 3500)), "shiftRpm": float(shift_rpm or (redline * 0.95 if redline else 7000)), "redlineRpm": float(redline or 7500), "idleRpm": float(idle_rpm or 800), "carName": car_screen_name or "Active Vehicle", } return { "CarPath": car_path, "CarScreenName": car_screen_name, "carScreenName": car_screen_name, "CarClassShortName": car_class_short_name, "carClassShortName": car_class_short_name, "TrackName": track_name, "trackName": track_name, "TrackDisplayName": track_display_name, "trackDisplayName": track_display_name, "TrackConfigName": track_config_name, "trackConfigName": track_config_name, "DriverName": driver_name, "driverName": driver_name, "CarNumber": car_number, "carNumber": car_number, "DriverCarSLFirstRPM": first_rpm, "DriverCarSLShiftRPM": shift_rpm, "DriverCarSLLastRPM": last_rpm, "DriverCarSLBlinkRPM": blink_rpm, "DriverCarRedLine": redline, "DriverCarIdleRPM": idle_rpm, "engineLimits": engine_limits, "engine_limits": engine_limits, } def extract_telemetry_packet(ir_instance: Any, car_info: Dict[str, Any] = None) -> Dict[str, Any]: """Extracts and formats the 60Hz telemetry frame from iRacing.""" if car_info is None: car_info = { "CarPath": "", "CarScreenName": "", "CarClassShortName": "", "TrackName": "", "TrackDisplayName": "", "TrackConfigName": "", "DriverName": "", "CarNumber": "", } # Speed in m/s -> MPH speed_ms = float(get_channel(ir_instance, 'Speed', 0.0)) speed_mph = round(speed_ms * 2.23694, 1) # Engine RPM & Gear rpm = round(float(get_channel(ir_instance, 'RPM', 0.0))) gear = int(get_channel(ir_instance, 'Gear', 0)) # Controls: Throttle, Brake, Clutch (0.0 - 1.0 -> 0 - 100%) throttle_pct = round(float(get_channel(ir_instance, 'Throttle', 0.0)) * 100, 1) brake_pct = round(float(get_channel(ir_instance, 'Brake', 0.0)) * 100, 1) clutch_pct = round(float(get_channel(ir_instance, 'Clutch', 0.0)) * 100, 1) steer_rad = round(float(get_channel(ir_instance, 'SteeringWheelAngle', 0.0)), 3) # G-Forces: m/s^2 -> G (1G = 9.80665 m/s^2) lat_accel = float(get_channel(ir_instance, 'LatAccel', 0.0)) long_accel = float(get_channel(ir_instance, 'LongAccel', 0.0)) vert_accel = float(get_channel(ir_instance, 'VertAccel', 0.0)) lateral_g = round(lat_accel / 9.80665, 2) longitudinal_g = round(long_accel / 9.80665, 2) vert_g = round(vert_accel / 9.80665, 2) # Shock Deflections (meters) fl_shock = round(float(get_channel(ir_instance, 'LFshockDefl', get_channel(ir_instance, 'LFshockDeflection', 0.0))), 4) fr_shock = round(float(get_channel(ir_instance, 'RFshockDefl', get_channel(ir_instance, 'RFshockDeflection', 0.0))), 4) rl_shock = round(float(get_channel(ir_instance, 'LRshockDefl', get_channel(ir_instance, 'LRshockDeflection', 0.0))), 4) rr_shock = round(float(get_channel(ir_instance, 'RRshockDefl', get_channel(ir_instance, 'RRshockDeflection', 0.0))), 4) # Shock Velocities (m/s) for transient kerb strike detection fl_shock_vel = abs(float(get_channel(ir_instance, 'LFshockVel', 0.0))) fr_shock_vel = abs(float(get_channel(ir_instance, 'RFshockVel', 0.0))) # Kerb / Rumble strip detection (strictly gated to moving vehicle > 8 mph to avoid pit box false strikes) lf_rumble = float(get_channel(ir_instance, 'TireLF_RumblePitch', 0.0)) rf_rumble = float(get_channel(ir_instance, 'TireRF_RumblePitch', 0.0)) if speed_mph >= 8.0: kerb_left = 100 if (lf_rumble > 0 or fl_shock_vel > 0.35) else 0 kerb_right = 100 if (rf_rumble > 0 or fr_shock_vel > 0.35) else 0 else: kerb_left = 0 kerb_right = 0 # Slip Ratios & ABS/TC fl_slip = round(float(get_channel(ir_instance, 'LFslipRatio', 0.0)), 3) fr_slip = round(float(get_channel(ir_instance, 'RFslipRatio', 0.0)), 3) rl_slip = round(float(get_channel(ir_instance, 'LRslipRatio', 0.0)), 3) rr_slip = round(float(get_channel(ir_instance, 'RRslipRatio', 0.0)), 3) ir_abs = bool(get_channel(ir_instance, 'BrakeABSactive', False)) abs_active = ir_abs or (brake_pct > 60 and (abs(fl_slip) > 0.18 or abs(fr_slip) > 0.18)) traction_active = (throttle_pct > 40 and (abs(rl_slip) > 0.22 or abs(rr_slip) > 0.22)) # Pit Limiter & Ignition Status for Smart Controls Cluster pit_limiter = bool(get_channel(ir_instance, 'PitSpeedLimiter', get_channel(ir_instance, 'PitLimiter', False))) volt = float(get_channel(ir_instance, 'Voltage', 0.0)) is_on_pit_road = bool(get_channel(ir_instance, 'OnPitRoad', False)) lap = int(get_channel(ir_instance, 'Lap', 1)) lap_completed = int(get_channel(ir_instance, 'LapCompleted', 0)) lap_current_lap_time = float(get_channel(ir_instance, 'LapCurrentLapTime', 0.0)) lap_last_lap_time = float(get_channel(ir_instance, 'LapLastLapTime', 0.0)) lap_best_lap_time = float(get_channel(ir_instance, 'LapBestLapTime', 0.0)) lap_dist_pct = float(get_channel(ir_instance, 'LapDistPct', 0.0)) yaw_rate = float(get_channel(ir_instance, 'YawRate', 0.0)) pitch_rate = float(get_channel(ir_instance, 'PitchRate', 0.0)) roll_rate = float(get_channel(ir_instance, 'RollRate', 0.0)) track_surface = int(get_channel(ir_instance, 'TrackSurface', 0)) incidents = int(get_channel(ir_instance, 'PlayerCarMyIncidentCount', 0)) # Zero out active driving physics if driver is out of car / in garage / replay (Zero fake 300 RPM!) is_on_track = bool(get_channel(ir_instance, 'IsOnTrack', False)) if not is_on_track: speed_mph = 0.0 rpm = 0 gear = 0 throttle_pct = 0.0 brake_pct = 0.0 clutch_pct = 0.0 lateral_g = 0.0 longitudinal_g = 0.0 vert_g = 0.0 kerb_left = 0 kerb_right = 0 engine_harmonics = 0 ignition_on = False else: # Engine harmonics: scaled relative to RPM (0 - 100%) engine_harmonics = min(100, max(0, int((rpm / 8500.0) * 100))) ignition_on = bool((rpm > 50) or (volt > 11.0) or (is_on_track and rpm > 0)) # Ingest ALL 300+ iRacing shared memory variables dynamically all_channels: Dict[str, Any] = {} if ir_instance and hasattr(ir_instance, '_var_headers_dict') and ir_instance._var_headers_dict: try: for key in ir_instance._var_headers_dict.keys(): try: val = ir_instance[key] all_channels[key] = sanitize_for_json(val) except Exception: pass except Exception: pass audio_devs = get_audio_output_devices() engine_limits = car_info.get("engineLimits") if (car_info and isinstance(car_info, dict)) else { "minRpm": 3500.0, "shiftRpm": 7000.0, "redlineRpm": 7500.0, "idleRpm": 800.0, "carName": "Active Vehicle", } return { "type": "telemetry", "source": "iracing", "bridge_version": BRIDGE_VERSION, "connected": True, "isOnTrack": is_on_track, "is_on_track": is_on_track, "isOnPitRoad": is_on_pit_road, "is_on_pit_road": is_on_pit_road, "lap": lap, "lapCompleted": lap_completed, "lap_completed": lap_completed, "lapCurrentLapTime": lap_current_lap_time, "lapLastLapTime": lap_last_lap_time, "lapBestLapTime": lap_best_lap_time, "lapDistPct": lap_dist_pct, "yawRate": yaw_rate, "pitchRate": pitch_rate, "rollRate": roll_rate, "trackSurface": track_surface, "incidents": incidents, "simulating": False, "simulationMode": "iracing", "car_info": car_info, "carInfo": car_info, "engineLimits": engine_limits, "engine_limits": engine_limits, "speedMph": speed_mph, "rpm": rpm, "gear": gear, "throttle": throttle_pct, "brake": brake_pct, "clutch": clutch_pct, "steer": steer_rad, "lateralG": lateral_g, "longitudinalG": longitudinal_g, "vertG": vert_g, "shockDeflection": { "fl": fl_shock, "fr": fr_shock, "rl": rl_shock, "rr": rr_shock, }, "slipRatio": { "fl": fl_slip, "fr": fr_slip, "rl": rl_slip, "rr": rr_slip, }, "absActive": abs_active, "tractionLossActive": traction_active, "kerbRumbleLeft": kerb_left, "kerbRumbleRight": kerb_right, "engineHarmonics": engine_harmonics, "all_channels": all_channels, "allChannels": all_channels, "audio_devices": audio_devs, "audioDevices": audio_devs, "esp32": ESP32_MGR.get_state(), "hardware": ESP32_MGR.get_state(), "boards": ESP32_MGR.get_boards_list(), "machine_passport": get_machine_passport(), "machinePassport": get_machine_passport(), "pitLimiterActive": pit_limiter, "pit_limiter": pit_limiter, "ignitionOn": ignition_on, "ignition_on": ignition_on, "timestamp": int(time.time() * 1000), } # ============================================================================ # PC MACHINE PASSPORT FINGERPRINTING (SMBIOS / WMI / CIM) # ============================================================================ MACHINE_PASSPORT_CACHE: Optional[Dict[str, Any]] = None def get_pc_machine_passport() -> Dict[str, Any]: """Retrieves or queries the persistent PC Machine Passport for the host rig.""" global MACHINE_PASSPORT_CACHE if MACHINE_PASSPORT_CACHE is not None: return MACHINE_PASSPORT_CACHE passport_path = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "machine_passport.json") if os.path.exists(passport_path): try: with open(passport_path, "r", encoding="utf-8") as f: cached = json.load(f) if isinstance(cached, dict) and cached.get("fingerprint"): pc_name = cached.get("pcName") or os.environ.get("COMPUTERNAME", "SimRig") clean_id = re.sub(r'[^a-zA-Z0-9]', '', pc_name).lower() or "simrig" short_fp = clean_id[-8:].upper() if len(clean_id) >= 8 else clean_id.upper().rjust(8, '0') if not cached.get("rigId"): cached["rigId"] = f"rig-{clean_id}" cached["rig_id"] = f"rig-{clean_id}" if not cached.get("name"): cached["name"] = f"{pc_name} Rig" cached["pcName"] = pc_name cached["pc_name"] = pc_name if not cached.get("fingerprint") or cached.get("fingerprint", "").startswith("GP-RIG-"): cached["hardwareFingerprint"] = cached.get("fingerprint") cached["fingerprint"] = f"PC-{short_fp}" MACHINE_PASSPORT_CACHE = cached return cached except Exception: pass pc_name = os.environ.get("COMPUTERNAME", "SimRig") mb_serial = "UNKNOWN" bios_uuid = "UNKNOWN" cpu_name = "x86_64 Processor" cpu_id = "UNKNOWN" drive_serial = "UNKNOWN" gpu_name = "Graphics Controller" if sys.platform == "win32": ps_code = ( "$mb = (Get-CimInstance Win32_BaseBoard -ErrorAction SilentlyContinue).SerialNumber; " "$bios = (Get-CimInstance Win32_ComputerSystemProduct -ErrorAction SilentlyContinue).UUID; " "$cpu = (Get-CimInstance Win32_Processor -ErrorAction SilentlyContinue).Name; " "$cpuId = (Get-CimInstance Win32_Processor -ErrorAction SilentlyContinue).ProcessorId; " "$drive = (Get-CimInstance Win32_DiskDrive -ErrorAction SilentlyContinue | Select-Object -First 1).SerialNumber; " "$gpus = Get-CimInstance Win32_VideoController -ErrorAction SilentlyContinue | Select-Object -ExpandProperty Name; " "$primaryGpu = $gpus | Where-Object { $_ -match 'NVIDIA|GeForce|RTX|GTX|Radeon|AMD' } | Select-Object -First 1; " "if (-not $primaryGpu) { $primaryGpu = $gpus | Where-Object { $_ -notmatch 'Virtual' -and $_ -notmatch 'Basic' -and $_ -notmatch 'Remote' } | Select-Object -First 1 }; " "if (-not $primaryGpu) { $primaryGpu = $gpus | Select-Object -First 1 }; " "$ramGb = [math]::Round(((Get-CimInstance Win32_PhysicalMemory -ErrorAction SilentlyContinue | Measure-Object -Property Capacity -Sum).Sum) / 1GB); " "[PSCustomObject]@{ mb = $mb; bios = $bios; cpu = $cpu; cpuId = $cpuId; drive = $drive; gpu = $primaryGpu; ramGb = $ramGb } | ConvertTo-Json -Compress" ) try: raw = subprocess.check_output( ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps_code], timeout=7, stderr=subprocess.DEVNULL ).decode("utf-8", errors="ignore").strip() data = json.loads(raw) mb_serial = str(data.get("mb") or "UNKNOWN").strip() bios_uuid = str(data.get("bios") or "UNKNOWN").strip() cpu_name = str(data.get("cpu") or "x86_64 Processor").strip() cpu_id = str(data.get("cpuId") or "UNKNOWN").strip() drive_serial = str(data.get("drive") or "UNKNOWN").strip() gpu_name = str(data.get("gpu") or "Graphics Controller").strip() ram_gb = data.get("ramGb") ram_str = f"{ram_gb}GB" if ram_gb else "" except Exception as e: print(f"[PASSPORT QUERY ERROR] {e}") seed = f"{mb_serial}:{bios_uuid}:{cpu_id}:{drive_serial}" fp = hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16].upper() bt_mac = "" if sys.platform == "win32": try: import winrt.windows.devices.bluetooth as bt adapter = bt.BluetoothAdapter.get_default_async().get() if adapter and adapter.bluetooth_address: raw_addr = adapter.bluetooth_address bt_mac = ":".join(f"{(raw_addr >> (i * 8)) & 0xFF:02X}" for i in reversed(range(6))) except Exception: pass clean_id = re.sub(r'[^a-zA-Z0-9]', '', pc_name).lower() or "simrig" short_fp = clean_id[-8:].upper() if len(clean_id) >= 8 else clean_id.upper().rjust(8, '0') MACHINE_PASSPORT_CACHE = { "fingerprint": f"PC-{short_fp}", "hardwareFingerprint": f"GP-RIG-{fp}", "pcName": pc_name, "pc_name": pc_name, "rigId": f"rig-{clean_id}", "rig_id": f"rig-{clean_id}", "name": f"{pc_name} Rig", "gpuName": gpu_name, "gpu_name": gpu_name, "cpuName": cpu_name, "cpu_name": cpu_name, "cpu": cpu_name, "ramGb": ram_gb, "ram_gb": ram_gb, "ram": ram_str, "motherboardSerial": mb_serial, "motherboard_serial": mb_serial, "biosUuid": bios_uuid, "bios_uuid": bios_uuid, "driveSerial": drive_serial, "drive_serial": drive_serial, "bluetoothMac": bt_mac, "bluetooth_mac": bt_mac, "verified": True, } try: os.makedirs(os.path.dirname(passport_path), exist_ok=True) with open(passport_path, "w", encoding="utf-8") as f: json.dump(MACHINE_PASSPORT_CACHE, f, indent=2) except Exception: pass return MACHINE_PASSPORT_CACHE def get_machine_passport() -> Dict[str, Any]: """Retrieves deterministic machine identity and hardware specs for the host rig.""" try: pc_name = os.environ.get("COMPUTERNAME", socket.gethostname()) except Exception: pc_name = os.environ.get("COMPUTERNAME", "SimRig") clean_id = re.sub(r'[^a-zA-Z0-9]', '', pc_name).lower() or "simrig" short_fp = clean_id[-8:].upper() if len(clean_id) >= 8 else clean_id.upper().rjust(8, '0') base_info = {} try: base_info = get_pc_machine_passport() or {} except Exception: pass passport = { **base_info, "pcName": pc_name, "pc_name": pc_name, "fingerprint": f"PC-{short_fp}", "rigId": f"rig-{clean_id}", "rig_id": f"rig-{clean_id}", "name": f"{pc_name} Rig", "verified": True, } return passport # ============================================================================ # SIMOS CLOUD PASSPORT & RIG SYNC ENGINE # ============================================================================ def generate_claim_qr_code(url: str) -> Optional[str]: """Generates and caches a QR code image locally for scanning from mobile.""" try: qr_dir = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS") os.makedirs(qr_dir, exist_ok=True) qr_path = os.path.join(qr_dir, "claim_qr.png") encoded_url = urllib.parse.quote(url, safe="") qr_service = f"https://api.qrserver.com/v1/create-qr-code/?size=250x250&data={encoded_url}" req = urllib.request.Request(qr_service, headers={"User-Agent": "GridPass-SimOS"}) with urllib.request.urlopen(req, timeout=5) as resp, open(qr_path, "wb") as out_f: out_f.write(resp.read()) return qr_path except Exception: return None def sync_to_gridpass(host_url: Optional[str] = None) -> Optional[Dict[str, Any]]: """ Directly synchronizes the host PC machine passport and rig hardware stack to GridPass Cloud. Target: http://localhost:3005/api/simos/sync-rig with fallback to https://gridpass.app/api/simos/sync-rig. """ config_dir = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS") config_path = os.path.join(config_dir, "config.json") saved_config: Dict[str, Any] = {} if os.path.exists(config_path): try: with open(config_path, "r", encoding="utf-8") as f: saved_config = json.load(f) except Exception: pass passport = get_pc_machine_passport() pairing_token = ( saved_config.get("pairing_token") or saved_config.get("pairingToken") or os.environ.get("GRIDPASS_PAIRING_TOKEN") ) rig_id = ( saved_config.get("rig_id") or saved_config.get("rigId") or passport.get("rigId") ) payload: Dict[str, Any] = { "fingerprint": passport.get("fingerprint"), "hardwareFingerprint": passport.get("hardwareFingerprint"), "rigId": rig_id, "name": saved_config.get("name") or passport.get("name"), "pcName": passport.get("pcName"), "gpuName": passport.get("gpuName"), "cpuName": passport.get("cpuName"), "ramGb": passport.get("ramGb"), "ram": passport.get("ram"), "displaySetup": saved_config.get("displaySetup") or "triples", "chassisBrand": saved_config.get("chassisBrand", ""), "wheelbaseBrand": saved_config.get("wheelbaseBrand", ""), "wheelbaseModel": saved_config.get("wheelbaseModel", ""), "pedalsBrand": saved_config.get("pedalsBrand", ""), "pedalsModel": saved_config.get("pedalsModel", ""), "hapticsSummary": saved_config.get("hapticsSummary", ""), } if pairing_token: payload["pairingToken"] = str(pairing_token).strip() endpoints = [] if host_url: endpoints.append(f"{host_url.rstrip('/')}/api/simos/sync-rig") else: endpoints.append("http://localhost:3005/api/simos/sync-rig") endpoints.append("https://gridpass.app/api/simos/sync-rig") payload_bytes = json.dumps(payload).encode("utf-8") last_err = None for ep in endpoints: try: req = urllib.request.Request( ep, data=payload_bytes, headers={ "Content-Type": "application/json", "User-Agent": f"SimOS-Bridge/{BRIDGE_VERSION}", }, method="POST" ) with urllib.request.urlopen(req, timeout=6) as resp: if resp.status in (200, 201): raw_res = resp.read().decode("utf-8") data = json.loads(raw_res) if data.get("success"): slug = data.get("slug") or rig_id rig_url = data.get("url") or f"https://gridpass.app/simrigs/{slug}" print(f"\n[GRIDPASS SYNC] Rig synced to {rig_url}") # Cache confirmed rigId and slug to config.json try: os.makedirs(config_dir, exist_ok=True) saved_config["rig_id"] = data.get("rigId") or rig_id saved_config["slug"] = slug saved_config["url"] = rig_url saved_config["last_synced_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) with open(config_path, "w", encoding="utf-8") as f: json.dump(saved_config, f, indent=2) except Exception: pass # Directive: Windows Toast Notification & QR Code qr_path = generate_claim_qr_code(rig_url) print("\n" + "="*68) print(" 🏎️ GRIDPASS SIM RIG PASSPORT SYNCED") print(f" 🔗 Live Rig Profile: {rig_url}") print(" 📱 Scan QR or click notification to customize your build sheet!") print("="*68 + "\n") # Send rich Windows Toast Notification if sys.platform == "win32" and HAS_WINOTIFY: try: toast_icon = qr_path if (qr_path and os.path.exists(qr_path)) else os.path.join(config_dir, "gridpass_icon_rounded.png") if not os.path.exists(toast_icon): toast_icon = r"C:\_Projects\Gridpass-v5\web\public\downloads\gridpass_icon_rounded.png" toast = Notification( app_id="GridPass SimOS", title="🏎️ SimOS: Your Rig Profile is Ready!", msg="Click to view and claim your rig build sheet on GridPass, or scan the QR code to edit from your phone.", icon=toast_icon if os.path.exists(toast_icon) else "", launch=rig_url ) toast.add_actions(label="Open Rig Profile", launch=rig_url) toast.show() except Exception: try: send_tray_notification("🏎️ SimOS: Rig Synced!", f"Rig profile ready: {rig_url}") except Exception: pass else: try: send_tray_notification("🏎️ SimOS: Rig Synced!", f"Rig profile ready: {rig_url}") except Exception: pass return data except Exception as e: last_err = e continue if last_err: print(f"⚠️ [GRIDPASS SYNC WARNING] Cloud sync deferred: {last_err}") return None sync_rig_to_gridpass = sync_to_gridpass # ============================================================================ # 4. UNIVERSAL HARDWARE CONTROLLER & MULTI-BOARD CHIP ID ROUTER # ============================================================================ FACTORY_S3_PINS = { "led_pin": 6, "fan_l": 7, "fan_r": 8, "brake": 4, "gas": 5, "onboard_rgb": 48, "relay": 9, } class UniversalHardwareManager: """Universal Hardware Controller & Multi-Board Chip ID Router. Features: - Multi-board USB CDC discovery & active connection routing - Silicon Chip ID reporting & dynamic handshake (?ID -> !ID,MODEL=...,CHIP=...*) - Dynamic RAM pin configuration storage & transmission ($CFG,FAN_L=...,FAN_R=...*) - 60Hz $OUT telemetry streaming with PWM calculation & WS2812B RGB color encoding - Automatic persistence to %LOCALAPPDATA%\\GridPass\\SimOS\\hardware_pins.json - Hardware watchdog recovery & port conflict detection """ def __init__(self): self.boards: Dict[str, Dict[str, Any]] = {} self.lock = threading.Lock() self.last_poll_time: float = 0.0 self.tx_counter: int = 0 self.baudrate: int = 115200 self.status: str = "DISCONNECTED" self.error_message: Optional[str] = None self.firmware_version: str = BRIDGE_VERSION self.needs_flash: bool = False self.last_notified_port: Optional[str] = None self.pins_file_path = os.path.join( os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "hardware_pins.json" ) self.saved_pins_cache: Dict[str, Dict[str, int]] = self.load_saved_pins() self.lighting_layers_file = os.path.join( os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "lighting_layers.json" ) self.lighting_profiles_file = os.path.join( os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "lighting_profiles.json" ) self.lighting_layers: List[Dict[str, Any]] = self.load_lighting_layers() self.lighting_profiles: Dict[str, Any] = {} self.load_lighting_profiles() self.last_engine_limits: Optional[Dict[str, Any]] = None self.alignment_active: bool = False self.alignment_timer: Optional[threading.Timer] = None self.alignment_hex: Optional[str] = None self.active_config: Dict[str, Any] = { "wind": { "enabled": True, "speedCurve": "progressive", "minIdleBreeze": 10, "maxFanSpeed": 100, "pitFanOverride": False, }, "pedals": { "enabled": True, "brakeAbsThreshold": 80, "throttleTractionThreshold": 75, "brakeIntensity": 85, "throttleIntensity": 80, }, "shiftLights": { "enabled": True, "autoSyncCar": True, "totalLedCount": 16, "revLights": {"startLed": 4, "endLed": 11, "reverse": False}, "spotterLeft": {"startLed": 0, "endLed": 3}, "spotterRight": {"startLed": 12, "endLed": 15}, "flagHalo": {"startLed": 0, "endLed": 15}, "maxRpm": 8000, "shiftPointRpm": 7200, "brightness": 80, "spotterLeftActive": False, "spotterRightActive": False, "activeFlag": "none", } } TARGET_VIDS = { 0x303A, # Espressif (ESP32-S2/S3/C3/C6) 0x2341, # Arduino (MKR 1010, Uno, Mega, Nano, Micro, Leonardo, Due, Nano 33 IoT) 0x2A03, # Arduino (vintage / dog hunter) 0x2E8A, # Raspberry Pi (RP2040, Pico, Pico W, RP2350) 0x16C0, # PJRC Teensy (3.x / 4.x) 0x0483, # STMicroelectronics (STM32 Nucleo / BlackPill) 0x1A86, # QinHeng CH340 / CH341 0x10C4, # Silicon Labs CP210x } EXCLUDED_VIDS = { 0x346E, # Moza Racing 0x0EB7, # Fanatec 0x16D0, # Simucube / Bodnar 0x044F, # Thrustmaster 0x046D, # Logitech } @property def port(self) -> Optional[str]: for b in self.boards.values(): if b.get("status") == "CONNECTED": return b.get("port") return None @port.setter def port(self, val: Optional[str]): pass @property def stint_engine(self) -> Any: return globals().get("STINT_ENGINE") @property def board_name(self) -> str: for b in self.boards.values(): if b.get("status") == "CONNECTED": return b.get("board_name", "Microcontroller") return "Microcontroller" @property def device_desc(self) -> str: for b in self.boards.values(): if b.get("status") == "CONNECTED": return b.get("device_desc", "") return "" def load_saved_pins(self) -> Dict[str, Dict[str, int]]: try: if os.path.exists(self.pins_file_path): with open(self.pins_file_path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict): return data except Exception as e: print(f"[PINS LOAD ERROR] {e}") return {} def save_pins(self): try: os.makedirs(os.path.dirname(self.pins_file_path), exist_ok=True) with open(self.pins_file_path, "w", encoding="utf-8") as f: json.dump(self.saved_pins_cache, f, indent=2) except Exception as e: print(f"[PINS SAVE ERROR] {e}") def load_lighting_layers(self) -> List[Dict[str, Any]]: """Loads lighting layers from %LOCALAPPDATA%\\GridPass\\SimOS\\lighting_layers.json.""" try: if os.path.exists(self.lighting_layers_file): with open(self.lighting_layers_file, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): return data except Exception as e: print(f"[LIGHTING LAYERS LOAD ERROR] {e}") return [] def update_lighting_layers(self, layers: List[Dict[str, Any]]) -> bool: """Updates active layer stack in RAM and persists to %LOCALAPPDATA%\\GridPass\\SimOS\\lighting_layers.json.""" try: self.lighting_layers = layers if isinstance(layers, list) else [] os.makedirs(os.path.dirname(self.lighting_layers_file), exist_ok=True) with open(self.lighting_layers_file, "w", encoding="utf-8") as f: json.dump(self.lighting_layers, f, indent=2) return True except Exception as e: print(f"[LIGHTING LAYERS SAVE ERROR] {e}") return False def load_lighting_profiles(self) -> List[Dict[str, Any]]: """Reads %LOCALAPPDATA%\\GridPass\\SimOS\\lighting_profiles.json into memory and returns profiles list.""" try: if os.path.exists(self.lighting_profiles_file): with open(self.lighting_profiles_file, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): self.lighting_profiles = {p.get("id", str(i)): p for i, p in enumerate(data) if isinstance(p, dict)} return data elif isinstance(data, dict): self.lighting_profiles = data return list(data.values()) except Exception as e: print(f"[LIGHTING PROFILES LOAD ERROR] {e}") return list(self.lighting_profiles.values()) if isinstance(self.lighting_profiles, dict) else [] def save_lighting_profile(self, profile: Any) -> bool: """Persists lighting profile(s) to %LOCALAPPDATA%\\GridPass\\SimOS\\lighting_profiles.json.""" try: os.makedirs(os.path.dirname(self.lighting_profiles_file), exist_ok=True) if isinstance(profile, list): self.lighting_profiles = {p.get("id", str(i)): p for i, p in enumerate(profile) if isinstance(p, dict)} to_save = profile elif isinstance(profile, dict): p_id = profile.get("id") or f"prof_light_{int(time.time() * 1000)}" profile["id"] = p_id self.lighting_profiles[p_id] = profile to_save = list(self.lighting_profiles.values()) else: return False with open(self.lighting_profiles_file, "w", encoding="utf-8") as f: json.dump(to_save, f, indent=2) return True except Exception as e: print(f"[LIGHTING PROFILE SAVE ERROR] {e}") return False def get_default_pins_for_model(self, model: str) -> Dict[str, int]: m = model.upper() if "S3" in m or "ESP32S3" in m or "ESP32-S3" in m or model.lower() == "esp32s3": pins = dict(FACTORY_S3_PINS) pins["led_count"] = 300 return pins elif "MKR" in m: return {"fan_l": 2, "fan_r": 3, "brake": 4, "gas": 5, "led_pin": 6, "led_count": 300} elif "ESP32" in m: pins = dict(FACTORY_S3_PINS) pins["led_count"] = 300 return pins elif "MICRO" in m or "LEONARDO" in m: return {"fan_l": 5, "fan_r": 6, "brake": 9, "gas": 10, "led_pin": 11, "led_count": 16} elif "MEGA" in m: return {"fan_l": 2, "fan_r": 3, "brake": 4, "gas": 5, "led_pin": 6, "led_count": 16} else: return {"fan_l": 3, "fan_r": 5, "brake": 6, "gas": 9, "led_pin": 10, "led_count": 16} def identify_board(self, vid: Optional[int], pid: Optional[int], desc: str) -> str: """Resolves human-readable board name from USB VID/PID.""" if not vid: return "Microcontroller" if vid in (0x2341, 0x2A03): arduino_pids = { 0x8054: "Arduino MKR WiFi 1010", 0x804E: "Arduino MKR Zero", 0x8053: "Arduino MKR WAN 1300", 0x8055: "Arduino MKR GSM 1400", 0x8057: "Arduino MKR NB 1500", 0x8052: "Arduino MKR FOX 1200", 0x8037: "Arduino Micro", 0x8036: "Arduino Leonardo", 0x0043: "Arduino Uno", 0x0243: "Arduino Uno", 0x0042: "Arduino Mega 2560", 0x0058: "Arduino Nano 33 BLE", 0x805A: "Arduino Nano 33 IoT", 0x0070: "Arduino Nano RP2040 Connect", 0x003E: "Arduino Due", } if pid and pid in arduino_pids: return arduino_pids[pid] return "Arduino Board" elif vid == 0x303A: return "ESP32-S3" elif vid == 0x2E8A: return "Raspberry Pi RP2040" elif vid == 0x16C0: return "Teensy 4.x" elif vid == 0x0483: return "STM32 Board" elif vid == 0x1A86: return "CH340 Microcontroller" elif vid == 0x10C4: return "CP2102 Microcontroller" return "Microcontroller" def update_config(self, cfg: Dict[str, Any]): """Update hardware config dynamically without requiring flash.""" if not isinstance(cfg, dict): return if "wind" in cfg and isinstance(cfg["wind"], dict): self.active_config["wind"].update(cfg["wind"]) if "pedals" in cfg and isinstance(cfg["pedals"], dict): self.active_config["pedals"].update(cfg["pedals"]) if "shiftLights" in cfg and isinstance(cfg["shiftLights"], dict): self.active_config["shiftLights"].update(cfg["shiftLights"]) def update_hardware_pins(self, pins: Dict[str, Any], target_chip_id: Optional[str] = None) -> bool: """Applies dynamic pin updates, persists to hardware_pins.json, and pushes $CFG down serial.""" with self.lock: target_board = None if target_chip_id and target_chip_id in self.boards: target_board = self.boards[target_chip_id] else: for b in self.boards.values(): if b.get("status") == "CONNECTED": target_board = b break is_s3 = False if target_board: tb_model = str(target_board.get("model", "")).upper() tb_type = str(target_board.get("board_type", "")).upper() is_s3 = any(x in tb_model for x in ["ESP32S3", "ESP32-S3", "S3"]) or "ESP32S3" in tb_type or "S3" in tb_type elif target_chip_id: is_s3 = "S3" in target_chip_id.upper() or "303A" in target_chip_id.upper() if is_s3: clean_pins = dict(FACTORY_S3_PINS) clean_pins["led_count"] = int(pins.get("led_count", pins.get("ledCount", 300))) else: clean_pins = { "fan_l": int(pins.get("fan_l", pins.get("fanLeft", 2))), "fan_r": int(pins.get("fan_r", pins.get("fanRight", 3))), "brake": int(pins.get("brake", pins.get("brakeRumble", 4))), "gas": int(pins.get("gas", pins.get("gasRumble", 5))), "led_pin": int(pins.get("led_pin", pins.get("ledDataPin", 6))), "led_count": int(pins.get("led_count", pins.get("ledCount", 16))), } cid = target_board["chip_id"] if target_board else (target_chip_id or "default") self.saved_pins_cache[cid] = clean_pins self.save_pins() if target_board: target_board["pins"] = clean_pins conn = target_board.get("serial_conn") if conn and getattr(conn, 'is_open', False): cfg_pkt = ( f"$CFG,FAN_L={clean_pins['fan_l']}," f"FAN_R={clean_pins['fan_r']}," f"BRK={clean_pins['brake']}," f"GAS={clean_pins['gas']}," f"LED_PIN={clean_pins['led_pin']}," f"LED_COUNT={clean_pins['led_count']}," f"BRIGHT=40*\n" ) try: conn.write(cfg_pkt.encode('utf-8')) time.sleep(0.03) black_leds = "000000" * clean_pins['led_count'] clear_pkt = f"$OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{black_leds}*\n" conn.write(clear_pkt.encode('utf-8')) print(f"⚡ [HARDWARE PIN CONFIG SENT] {target_board['board_name']} [{cid}]: {cfg_pkt.strip()}") return True except Exception as e: print(f"⚠️ [CFG SEND ERROR] {e}") return False return True def find_esp32_port(self) -> Optional[Tuple[str, str, str]]: for b in self.boards.values(): if b.get("status") == "CONNECTED": return (b["port"], b.get("device_desc", ""), b.get("board_name", "Microcontroller")) return None def poll_connection(self): """Polls all USB serial ports, manages multiple connected boards, and discovers chip IDs.""" if not HAS_SERIAL: return if self.status == "FLASHING": return now = time.time() if now - self.last_poll_time < 1.0: return self.last_poll_time = now try: ports = list(serial.tools.list_ports.comports()) except Exception: return with self.lock: # Check existing connections available_port_names = [p.device.upper() for p in ports] disconnected_ids = [] for chip_id, b in list(self.boards.items()): conn = b.get("serial_conn") port_name = (b.get("port") or "").upper() if not conn or not getattr(conn, 'is_open', False) or (port_name and port_name not in available_port_names): disconnected_ids.append(chip_id) for cid in disconnected_ids: try: conn = self.boards[cid].get("serial_conn") if conn: conn.close() except Exception: pass del self.boards[cid] active_conns = [b for b in self.boards.values() if b.get('status') == 'CONNECTED'] if not active_conns and self.status != "FLASHING": self.status = "DISCONNECTED" self.error_message = None # Scan available ports for p in ports: vid = getattr(p, 'vid', None) pid = getattr(p, 'pid', None) desc = p.description or '' hwid = p.hwid or '' mfg = getattr(p, 'manufacturer', '') or '' combined = f"{desc} {hwid} {mfg}".lower() if vid and vid in self.EXCLUDED_VIDS: continue if any(x in combined for x in ['moza', 'fanatec', 'simucube', 'thrustmaster', 'logitech']): continue matched = False board_name = "Microcontroller" model = "ARDUINO" if vid and vid in self.TARGET_VIDS: matched = True board_name = self.identify_board(vid, pid, desc) if vid == 0x2341 and pid == 0x8054: model = "MKR1010" elif vid == 0x303A: model = "ESP32S3" elif vid == 0x2341 and pid == 0x8037: model = "MICRO" elif vid == 0x2341 and pid in (0x0043, 0x0243): model = "UNO" elif vid == 0x2341 and pid == 0x0042: model = "MEGA2560" else: strict_identifiers = ['espressif', '0x303a', 'usb-enhanced-serial', 'ch340', 'ch341', 'cp210', 'esp32', 'rp2040', 'teensy', 'arduino'] if any(ident in combined for ident in strict_identifiers): matched = True board_name = "Microcontroller" if not matched: continue # Determine chip_id serial_num = getattr(p, 'serial_number', None) if not serial_num and "SER=" in hwid: try: serial_num = hwid.split("SER=")[1].split(" ")[0].strip() except Exception: pass chip_id = "" if serial_num: chip_id = f"{model[:3]}-{serial_num[:8]}" else: chip_id = f"{model[:3]}-{p.device}" # Check if already connected on this port already_active = False for existing_cid, eb in self.boards.items(): if eb.get("port") == p.device and eb.get("status") == "CONNECTED" and eb.get("serial_conn"): already_active = True break if already_active: continue # Attempt to open serial connection try: ser = serial.Serial( port=p.device, baudrate=self.baudrate, timeout=0.15, write_timeout=0.3 ) time.sleep(0.1) # Send $WHOAMI and ?ID to query daemon identity fw_version = None try: ser.write(b"$WHOAMI*\n") ser.write(b"?ID\n") time.sleep(0.08) for _ in range(5): if getattr(ser, 'in_waiting', 0) > 0: resp = ser.readline().decode('utf-8', errors='replace').strip() if resp.startswith("$IDENT,") and resp.endswith("*"): parts = resp[7:-1].split(",") for part in parts: if part.startswith("BOARD="): model = part.split("=")[1] elif part.startswith("CHIP="): chip_id = part.split("=")[1] elif part.startswith("FW="): fw_version = part.split("=")[1] break elif resp.startswith("!ID,") and resp.endswith("*"): parts = resp[4:-1].split(",") for part in parts: if part.startswith("MODEL="): model = part.split("=")[1] elif part.startswith("CHIP="): chip_id = part.split("=")[1] elif part.startswith("VER="): fw_version = part.split("=")[1] break time.sleep(0.02) except Exception: pass TARGET_FW = "1.3.0" update_required = (fw_version != TARGET_FW) # Load pins is_s3 = any(x in model.upper() for x in ["ESP32S3", "ESP32-S3", "S3"]) or model.lower() == "esp32s3" if is_s3: pins = dict(FACTORY_S3_PINS) pins["led_count"] = self.saved_pins_cache.get(chip_id, {}).get("led_count", 300) self.saved_pins_cache[chip_id] = pins self.save_pins() else: if chip_id not in self.saved_pins_cache: self.saved_pins_cache[chip_id] = self.get_default_pins_for_model(model) self.save_pins() pins = self.saved_pins_cache[chip_id] if "MKR" in model.upper() and pins.get("led_count", 0) < 300: pins["led_count"] = 300 self.save_pins() led_cnt = max(8, min(300, int(pins.get("led_count", 300)))) # Push $CFG packet immediately cfg_packet = ( f"$CFG,FAN_L={pins.get('fan_l', FACTORY_S3_PINS['fan_l'])}," f"FAN_R={pins.get('fan_r', FACTORY_S3_PINS['fan_r'])}," f"BRK={pins.get('brake', FACTORY_S3_PINS['brake'])}," f"GAS={pins.get('gas', FACTORY_S3_PINS['gas'])}," f"LED_PIN={pins.get('led_pin', FACTORY_S3_PINS['led_pin'])}," f"LED_COUNT={led_cnt}," f"BRIGHT=40*\n" ) ser.write(cfg_packet.encode('utf-8')) time.sleep(0.04) # Immediately send an all-black $OUT packet to instantly clear any power-on rainbow artifacts! black_leds = "000000" * led_cnt clear_pkt = f"$OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{black_leds}*\n" ser.write(clear_pkt.encode('utf-8')) self.boards[chip_id] = { "chip_id": chip_id, "port": p.device, "board_name": board_name, "model": model, "board_type": model.lower(), "device_desc": desc, "serial_conn": ser, "status": "CONNECTED", "error": None, "role": "primary" if len([b for b in self.boards.values() if b.get('status') == 'CONNECTED']) == 0 else "aux", "pins": pins, "firmware_version": fw_version, "target_version": TARGET_FW, "update_required": update_required, "last_tx_time": time.time(), } self.status = "CONNECTED" self.error_message = None fw_tag = f"v{fw_version}" if fw_version else "unflashed" req_tag = " [UPDATE REQUIRED]" if update_required else " [UP TO DATE]" print(f"🟢 [HARDWARE CONNECTED] {board_name} [{chip_id}] on {p.device} · FW: {fw_tag}{req_tag} · Config sent!") if "ESP32S3" in model.upper() or "ESP32-S3" in model.upper(): try: if "IR_INJECTOR" in globals(): inj_res = IR_INJECTOR.auto_inject_on_connect() if inj_res.get("injected"): print("🎮 [IR_INJECTOR] Auto-configured iRacing button bindings for ESP32-S3 controller.") except Exception as inj_err: print(f"⚠️ [IR_INJECTOR] Auto-inject on connect error: {inj_err}") if p.device != self.last_notified_port: self.last_notified_port = p.device send_tray_notification( "🔌 Hardware Connected", f"{board_name} [{chip_id}] on {p.device}" ) except (serial.SerialException, PermissionError) as se: err_str = str(se) if "PermissionError" in err_str or "Access is denied" in err_str or "could not open port" in err_str.lower(): self.status = "CONFLICT" self.error_message = f"{p.device} is locked by another application. Please close open serial monitors." else: self.status = "ERROR" self.error_message = f"Failed to open {p.device}: {err_str}" except Exception as e: self.status = "ERROR" self.error_message = str(e) # Update overall status connected_boards = [b for b in self.boards.values() if b.get("status") == "CONNECTED"] if connected_boards: self.status = "CONNECTED" self.error_message = None elif self.status != "CONFLICT": self.status = "DISCONNECTED" def send_packet(self, packet_str: str) -> bool: sent = False with self.lock: if not packet_str.endswith('\n'): packet_str += '\n' for b in self.boards.values(): conn = b.get("serial_conn") if conn and getattr(conn, 'is_open', False): try: conn.write(packet_str.encode('utf-8')) sent = True b["last_tx_time"] = time.time() except Exception: pass return sent def close_serial(self): self.status = "FLASHING" with self.lock: for b in self.boards.values(): conn = b.get("serial_conn") if conn: try: conn.close() except Exception: pass b["serial_conn"] = None b["status"] = "DISCONNECTED" time.sleep(0.3) def get_boards_list(self) -> List[Dict[str, Any]]: result = [] for b in self.boards.values(): pins = b.get("pins", {}) fw = b.get("firmware_version") target_fw = b.get("target_version", "1.3.0") upd_req = b.get("update_required", True) result.append({ "chipId": b["chip_id"], "port": b["port"], "boardName": b["board_name"], "model": b.get("model", ""), "boardType": b.get("board_type", (b.get("model") or "mkr1010").lower()), "role": b.get("role", "primary"), "status": b.get("status", "DISCONNECTED"), "error": b.get("error"), "firmwareVersion": fw, "targetVersion": target_fw, "updateRequired": upd_req, "pins": { "fanLeft": pins.get("fan_l", 2), "fanRight": pins.get("fan_r", 3), "brakeRumble": pins.get("brake", 4), "gasRumble": pins.get("gas", 5), "ledDataPin": pins.get("led_pin", 6), "ledCount": pins.get("led_count", 16), "onboardRgb": pins.get("onboard_rgb", 48), "relayPin": pins.get("relay", 9), } }) return result def get_state(self) -> Dict[str, Any]: primary_b = None for b in self.boards.values(): if b.get("status") == "CONNECTED": primary_b = b break port_str = primary_b["port"] if primary_b else "" board_name = primary_b["board_name"] if primary_b else "Microcontroller" chip_id = primary_b["chip_id"] if primary_b else "" pins = primary_b["pins"] if primary_b else {} fw_version = primary_b.get("firmware_version") if primary_b else None target_fw = primary_b.get("target_version", "1.3.0") if primary_b else "1.3.0" upd_required = primary_b.get("update_required", True) if primary_b else False board_type = (primary_b.get("model", "mkr1010") if primary_b else "mkr1010").lower() return { "connected": len([b for b in self.boards.values() if b.get("status") == "CONNECTED"]) > 0, "port": port_str, "deviceDesc": primary_b.get("device_desc", "") if primary_b else "", "boardName": board_name, "chipId": chip_id, "status": self.status, "error": self.error_message, "firmwareVersion": fw_version, "firmware_version": fw_version, "targetVersion": target_fw, "target_version": target_fw, "updateRequired": upd_required, "update_required": upd_required, "needsFlash": upd_required, "needs_flash": upd_required, "baudRate": self.baudrate, "pins": pins, "board": { "connected": primary_b is not None, "port": port_str, "board_type": board_type, "boardType": board_type, "board_name": board_name, "chip_id": chip_id, "chipId": chip_id, "firmware_version": fw_version, "firmwareVersion": fw_version, "target_version": target_fw, "targetVersion": target_fw, "update_required": upd_required, "updateRequired": upd_required, "pins": pins, }, "boards": self.get_boards_list(), "activeChipId": chip_id, "stint": STINT_ENGINE.get_telemetry_dict() if "STINT_ENGINE" in globals() and STINT_ENGINE else None, } @staticmethod def apply_brightness(hex_color: str, brightness: float = 100.0) -> str: """Scales an RGB hex color by brightness percentage (0-100).""" if not hex_color: return "000000" c = str(hex_color).lstrip("#").strip() if len(c) != 6: c = "FFFFFF" try: r = int(c[0:2], 16) g = int(c[2:4], 16) b = int(c[4:6], 16) except ValueError: r, g, b = 255, 255, 255 scale = max(0.0, min(1.0, float(brightness) / 100.0)) r = int(round(r * scale)) g = int(round(g * scale)) b = int(round(b * scale)) return f"{r:02X}{g:02X}{b:02X}" def build_rev_ladder_segment( self, rpm: int, start_led: int, end_led: int, reverse: bool = False, color_scheme: str = "classic_f1", brightness: float = 100.0, engine_limits: Optional[Dict[str, Any]] = None ) -> Dict[int, str]: """Calculates WS2812B RGB colors for an RPM ladder segment according to vehicle limits & chosen color scheme.""" sl = self.active_config.get("shiftLights", {}) limits = engine_limits or self.last_engine_limits auto_sync = sl.get("autoSyncCar", True) if auto_sync and limits and isinstance(limits, dict): shift_rpm = int(limits.get("shiftRpm") or sl.get("shiftPointRpm", 7200)) max_rpm = int(limits.get("redlineRpm") or sl.get("maxRpm", 8000)) min_rpm = int(limits.get("minRpm") or (max_rpm * 0.55)) else: shift_rpm = int(sl.get("shiftPointRpm", 7200)) max_rpm = int(sl.get("maxRpm", 8000)) min_rpm = int(sl.get("minRpm") or (max_rpm * 0.55)) low = min(start_led, end_led) high = max(start_led, end_led) segment_len = max(1, high - low + 1) res: Dict[int, str] = {} for i in range(low, high + 1): res[i] = "000000" # Shift point flash strobe if rpm >= shift_rpm: strobe_on = (int(time.time() * 1000) % 120 < 60) if strobe_on: if color_scheme == "porsche_gt3": strobe_color = "0064FF" elif color_scheme == "indycar": strobe_color = "FF00FF" elif color_scheme == "v8_stockcar": strobe_color = "FF0000" else: strobe_color = "0064FF" strobe_hex = self.apply_brightness(strobe_color, brightness) for i in range(low, high + 1): res[i] = strobe_hex return res # Normal RPM progression sweep if rpm > min_rpm and shift_rpm > min_rpm: frac = min(1.0, max(0.0, (rpm - min_rpm) / float(shift_rpm - min_rpm))) active_count = int(round(frac * segment_len)) is_rev = reverse if (start_led <= end_led) else (not reverse) for pos in range(segment_len): idx = (high - pos) if is_rev else (low + pos) if pos < active_count: ratio = pos / float(segment_len) if color_scheme == "porsche_gt3": raw_col = "FFAA00" if ratio < 0.65 else "0066FF" elif color_scheme == "indycar": if ratio < 0.35: raw_col = "00FF00" elif ratio < 0.65: raw_col = "FFFF00" else: raw_col = "FF0000" elif color_scheme == "v8_stockcar": if ratio < 0.40: raw_col = "FFFF00" elif ratio < 0.75: raw_col = "FF8800" else: raw_col = "FF0000" else: # Classic F1 if ratio < 0.35: raw_col = "00FF00" elif ratio < 0.70: raw_col = "FFFF00" elif ratio < 0.90: raw_col = "FF4500" else: raw_col = "FF0000" res[idx] = self.apply_brightness(raw_col, brightness) return res def composite_lighting_frame(self, telemetry_data: Dict[str, Any], led_count: int) -> str: """60Hz Multi-Layer Lighting Compositor. Initializes frame buffer, sorts active enabled layers by priority ascending (lowest draws first, highest on top), evaluates trigger conditions against telemetry, and paints WS2812B frame buffer. """ colors = ["000000"] * led_count if not self.lighting_layers: return "".join(colors) active_layers = [l for l in self.lighting_layers if isinstance(l, dict) and l.get("enabled", True)] active_layers.sort(key=lambda l: int(l.get("priority", 50))) is_on_track = bool(telemetry_data.get("isOnTrack", telemetry_data.get("is_on_track", False))) speed_mph = float(telemetry_data.get("speedMph", telemetry_data.get("speed_mph", telemetry_data.get("speed", 0.0)))) off_track = bool(telemetry_data.get("offTrack", telemetry_data.get("off_track", False))) kerb_l = float(telemetry_data.get("kerbRumbleLeft", telemetry_data.get("kerb_rumble_left", 0.0))) kerb_r = float(telemetry_data.get("kerbRumbleRight", telemetry_data.get("kerb_rumble_right", 0.0))) ignition_on = bool(telemetry_data.get("ignitionOn", telemetry_data.get("ignition_on", True))) pit_limiter = bool(telemetry_data.get("pitLimiterActive", telemetry_data.get("pit_limiter_active", False))) rpm = int(round(float(telemetry_data.get("rpm", 0)))) sl = self.active_config.get("shiftLights", {}) pedals = self.active_config.get("pedals", {}) for layer in active_layers: trigger = layer.get("trigger", "always") condition_met = False if trigger == "always" or not trigger: condition_met = True elif trigger == "on_track": condition_met = is_on_track elif trigger == "in_garage": condition_met = not is_on_track elif trigger == "car_stopped": condition_met = (speed_mph < 0.5) elif trigger == "off_track": condition_met = off_track elif trigger == "kerb_rumble": condition_met = (kerb_l > 40 or kerb_r > 40) elif trigger == "ignition_off": condition_met = not ignition_on elif trigger == "pit_limiter": condition_met = pit_limiter else: condition_met = True if not condition_met: continue l_type = layer.get("type", "") start_led = int(layer.get("startLed", layer.get("start_led", 0))) end_led = int(layer.get("endLed", layer.get("end_led", 0))) low_led = max(0, min(start_led, end_led)) high_led = min(led_count - 1, max(start_led, end_led)) brightness = float(layer.get("brightness", 100)) if l_type == "custom_color": hex_color = layer.get("color", "#FFFFFF") scaled_hex = self.apply_brightness(hex_color, brightness) for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex elif l_type == "rpm_ladder": reverse = bool(layer.get("reverse", False)) color_scheme = str(layer.get("colorScheme", layer.get("color_scheme", "classic_f1"))) engine_limits = telemetry_data.get("engineLimits") or telemetry_data.get("engine_limits") or self.last_engine_limits ladder_colors = self.build_rev_ladder_segment( rpm, start_led, end_led, reverse, color_scheme, brightness, engine_limits ) for idx, col in ladder_colors.items(): if 0 <= idx < led_count: colors[idx] = col elif l_type in ["spotter_left", "spotter_right"]: spot_l_active = bool(sl.get("spotterLeftActive", False) or kerb_l > 50 or telemetry_data.get("spotterLeftActive") or telemetry_data.get("spotter_left")) spot_r_active = bool(sl.get("spotterRightActive", False) or kerb_r > 50 or telemetry_data.get("spotterRightActive") or telemetry_data.get("spotter_right")) color_hex = layer.get("color", "#FF8800") scaled_hex = self.apply_brightness(color_hex, brightness) if l_type == "spotter_left" and spot_l_active: for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex elif l_type == "spotter_right" and spot_r_active: for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex elif l_type == "safety_flags": raw_flag = str(sl.get("activeFlag", telemetry_data.get("trackFlag", telemetry_data.get("flag", "none")))).lower() if raw_flag in ["yellow", "red", "blue", "green"]: if raw_flag == "yellow": flag_col = "FFFF00" if (int(time.time() * 1000) % 400 < 200) else "000000" elif raw_flag == "red": flag_col = "FF0000" elif raw_flag == "blue": flag_col = "0000FF" elif raw_flag == "green": flag_col = "00FF00" else: flag_col = "000000" scaled_hex = self.apply_brightness(flag_col, brightness) for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex elif l_type == "abs_lockup": brake_pct = float(telemetry_data.get("brake", 0)) abs_thresh = float(pedals.get("brakeAbsThreshold", 80)) abs_active = bool(telemetry_data.get("absActive", telemetry_data.get("abs_active", False))) or (brake_pct >= abs_thresh) if abs_active: # Strobe Red/Magenta strobe_col = "FF0000" if (int(time.time() * 1000) % 120 < 60) else "FF00FF" scaled_hex = self.apply_brightness(strobe_col, brightness) for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex elif l_type == "traction_control": throttle_pct = float(telemetry_data.get("throttle", 0)) tc_thresh = float(pedals.get("throttleTractionThreshold", 75)) tc_active = bool(telemetry_data.get("tractionLossActive", telemetry_data.get("traction_loss_active", False))) or (throttle_pct >= tc_thresh) if tc_active: # Strobe Yellow strobe_col = "FFFF00" if (int(time.time() * 1000) % 120 < 60) else "000000" scaled_hex = self.apply_brightness(strobe_col, brightness) for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex elif l_type == "ambient": color_hex = layer.get("color", "#00E5FF") scaled_hex = self.apply_brightness(color_hex, brightness) for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex elif l_type == "kerb_rumble": if kerb_l > 40 or kerb_r > 40: color_hex = layer.get("color", "#00FFFF") scaled_hex = self.apply_brightness(color_hex, brightness) for i in range(low_led, high_led + 1): if 0 <= i < led_count: colors[i] = scaled_hex return "".join(colors) def build_led_hex_string(self, rpm: int, flag: str, spot_l: int, spot_r: int, led_count: int, engine_limits: Optional[Dict[str, Any]] = None) -> str: sl = self.active_config.get("shiftLights", {}) limits = engine_limits or self.last_engine_limits auto_sync = sl.get("autoSyncCar", True) if auto_sync and limits and isinstance(limits, dict): shift_rpm = int(limits.get("shiftRpm") or sl.get("shiftPointRpm", 7200)) max_rpm = int(limits.get("redlineRpm") or sl.get("maxRpm", 8000)) min_rpm = int(limits.get("minRpm") or (max_rpm * 0.55)) else: shift_rpm = int(sl.get("shiftPointRpm", 7200)) max_rpm = int(sl.get("maxRpm", 8000)) min_rpm = int(sl.get("minRpm") or (max_rpm * 0.55)) colors = ["000000"] * led_count if flag in ["yellow", "red", "blue", "green"]: if flag == "yellow": color = "FFFF00" if (int(time.time() * 1000) % 400 < 200) else "000000" elif flag == "red": color = "FF0000" elif flag == "blue": color = "0000FF" else: color = "00FF00" colors = [color] * led_count else: if rpm >= shift_rpm: strobe = "0064FF" if (int(time.time() * 1000) % 120 < 60) else "000000" colors = [strobe] * led_count elif rpm > min_rpm and shift_rpm > min_rpm: frac = min(1.0, max(0.0, (rpm - min_rpm) / float(shift_rpm - min_rpm))) active_count = int(round(frac * led_count)) for i in range(min(active_count, led_count)): ratio = i / float(led_count) if ratio < 0.35: colors[i] = "00FF00" elif ratio < 0.70: colors[i] = "FFFF00" elif ratio < 0.90: colors[i] = "FF4500" else: colors[i] = "FF0000" if spot_l and led_count >= 2: colors[0] = "FFA500" colors[1] = "FFA500" if spot_r and led_count >= 2: colors[-1] = "FFA500" colors[-2] = "FFA500" return "".join(colors) def compute_and_send_telemetry(self, telemetry_data: Dict[str, Any]): """Builds $OUT,FL:...,FR:...,BRK:...,GAS:...,LEDS:...* and legacy $SIM,...* and streams down serial.""" connected_boards = [b for b in self.boards.values() if b.get("status") == "CONNECTED" and b.get("serial_conn")] if not connected_boards: return engine_limits = telemetry_data.get("engineLimits") or telemetry_data.get("engine_limits") if engine_limits and isinstance(engine_limits, dict): self.last_engine_limits = engine_limits speed = int(round(float(telemetry_data.get("speedMph", 0)))) rpm = int(round(float(telemetry_data.get("rpm", 0)))) gear = int(telemetry_data.get("gear", 0)) is_on_track = bool(telemetry_data.get("isOnTrack", telemetry_data.get("is_on_track", False))) # 1. Fan PWM fan_pwm = 0 wind = self.active_config.get("wind", {}) if wind.get("enabled", True): if wind.get("pitFanOverride", False): fan_pwm = 255 else: min_idle = float(wind.get("minIdleBreeze", 10)) max_fan = float(wind.get("maxFanSpeed", 100)) speed_curve = wind.get("speedCurve", "progressive") speed_frac = min(1.0, max(0.0, speed / 200.0)) if speed_curve == "progressive": curve_frac = math.pow(speed_frac, 1.6) elif speed_curve == "aggressive": curve_frac = math.pow(speed_frac, 0.8) else: curve_frac = speed_frac raw_pct = min_idle + curve_frac * (max_fan - min_idle) bounded_pct = min(max_fan, max(min_idle, raw_pct)) fan_pwm = int(round((bounded_pct / 100.0) * 255)) # 2. Pedals PWM pedals = self.active_config.get("pedals", {}) brake_pwm = 0 gas_pwm = 0 if pedals.get("enabled", True): brake_pct = float(telemetry_data.get("brake", 0)) throttle_pct = float(telemetry_data.get("throttle", 0)) abs_active = bool(telemetry_data.get("absActive", False)) or (brake_pct >= float(pedals.get("brakeAbsThreshold", 80))) tc_active = bool(telemetry_data.get("tractionLossActive", False)) or (throttle_pct >= float(pedals.get("throttleTractionThreshold", 75))) if abs_active: brake_pwm = int(round((float(pedals.get("brakeIntensity", 85)) / 100.0) * 255)) if tc_active: gas_pwm = int(round((float(pedals.get("throttleIntensity", 80)) / 100.0) * 255)) # 3. Spotter sl = self.active_config.get("shiftLights", {}) kerb_l = float(telemetry_data.get("kerbRumbleLeft", 0)) kerb_r = float(telemetry_data.get("kerbRumbleRight", 0)) spot_l = 1 if (sl.get("spotterLeftActive", False) or kerb_l > 50) else 0 spot_r = 1 if (sl.get("spotterRightActive", False) or kerb_r > 50) else 0 # 4. Track Flag active_flag = str(sl.get("activeFlag", "none")).lower() flag_char = "N" if active_flag == "green": flag_char = "G" elif active_flag == "yellow": flag_char = "Y" elif active_flag == "blue": flag_char = "B" elif active_flag == "red": flag_char = "R" legacy_packet = f"$SIM,{speed},{rpm},{gear},{fan_pwm},{brake_pwm},{gas_pwm},{spot_l},{spot_r},{flag_char}*\n" for b in connected_boards: conn = b.get("serial_conn") if not conn: continue pins = b.get("pins", {}) layer_max = max([int(l.get('endLed', l.get('end_led', 0))) + 1 for l in self.lighting_layers if isinstance(l, dict)] + [0]) led_count = max(300, int(pins.get("led_count", 300)), layer_max) if self.alignment_active and self.alignment_hex: led_hex = self.alignment_hex elif self.lighting_layers and len(self.lighting_layers) > 0: led_hex = self.composite_lighting_frame(telemetry_data, led_count) elif not is_on_track: led_hex = "000000" * led_count else: led_hex = self.build_led_hex_string(rpm, active_flag, spot_l, spot_r, led_count, engine_limits) out_packet = f"$OUT,FL:{fan_pwm},FR:{fan_pwm},BRK:{brake_pwm},GAS:{gas_pwm},LEDS:{led_hex}*\n" try: conn.write(out_packet.encode('utf-8')) conn.write(legacy_packet.encode('utf-8')) b["last_tx_time"] = time.time() self.tx_counter += 1 except Exception as e: print(f"⚠️ [TX ERROR on {b.get('port')}] {e}") b["status"] = "DISCONNECTED" try: conn.close() except Exception: pass b["serial_conn"] = None def send_test_rpm(self, rpm: int): """Builds $OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{hex}*\n based on current shift light config and writes to all connected boards!""" with self.lock: for b in self.boards.values(): if b.get("status") != "CONNECTED": continue conn = b.get("serial_conn") if not conn or not getattr(conn, 'is_open', False): continue pins = b.get("pins", {}) layer_max = max([int(l.get('endLed', l.get('end_led', 0))) + 1 for l in self.lighting_layers if isinstance(l, dict)] + [0]) led_count = max(300, int(pins.get("led_count", 300)), layer_max) if rpm <= 0: led_hex = "000000" * led_count elif self.lighting_layers and len(self.lighting_layers) > 0: mock_telem = { "rpm": rpm, "isOnTrack": True, "is_on_track": True, "speedMph": 60, "ignitionOn": True, "engineLimits": self.last_engine_limits } led_hex = self.composite_lighting_frame(mock_telem, led_count) else: led_hex = self.build_led_hex_string(rpm, "none", 0, 0, led_count, self.last_engine_limits) out_packet = f"$OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{led_hex}*\n" try: conn.write(out_packet.encode('utf-8')) b["last_tx_time"] = time.time() self.tx_counter += 1 except Exception as e: print(f"⚠️ [TEST RPM TX ERROR on {b.get('port')}] {e}") def render_alignment_frame(self, start_idx: int, end_idx: int, active_target: str = 'both'): """Builds a 300-LED hex string where: - Index start_idx = FF007F (Pink) - Index end_idx = FF8800 (Orange) - Indices between start_idx and end_idx = 003344 (Soft cyan/white guide) - All other indices = 000000 Sends $OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{hex}* to serial! Timeout watchdog automatically clears alignment frame 3.0s after the last nudge. """ self.alignment_active = True if self.alignment_timer is not None: try: self.alignment_timer.cancel() except Exception: pass def _timeout_clear(): self.alignment_active = False self.alignment_hex = None self.clear_leds() self.alignment_timer = threading.Timer(3.0, _timeout_clear) self.alignment_timer.daemon = True self.alignment_timer.start() with self.lock: for b in self.boards.values(): if b.get("status") != "CONNECTED": continue conn = b.get("serial_conn") if not conn or not getattr(conn, 'is_open', False): continue pins = b.get("pins", {}) layer_max = max([int(start_idx) + 1, int(end_idx) + 1] + [int(l.get('endLed', l.get('end_led', 0))) + 1 for l in self.lighting_layers if isinstance(l, dict)]) led_count = max(300, int(pins.get("led_count", 300)), layer_max) colors = ["000000"] * led_count s = max(0, min(led_count - 1, int(start_idx))) e = max(0, min(led_count - 1, int(end_idx))) low = min(s, e) high = max(s, e) for i in range(low + 1, high): colors[i] = "003344" if 0 <= s < led_count: colors[s] = "FF007F" if 0 <= e < led_count: colors[e] = "FF8800" if s == e and 0 <= s < led_count: colors[s] = "FF007F" if active_target == "start" else "FF8800" hex_str = "".join(colors) self.alignment_hex = hex_str out_packet = f"$OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{hex_str}*\n" try: conn.write(out_packet.encode('utf-8')) b["last_tx_time"] = time.time() self.tx_counter += 1 except Exception as e: print(f"⚠️ [ALIGNMENT TX ERROR on {b.get('port')}] {e}") def clear_leds(self): """Sends an all-black LED packet to turn off all LEDs on all connected boards.""" with self.lock: for b in self.boards.values(): if b.get("status") != "CONNECTED": continue conn = b.get("serial_conn") if not conn or not getattr(conn, 'is_open', False): continue pins = b.get("pins", {}) led_count = max(8, min(300, int(pins.get("led_count", 300)))) black_leds = "000000" * led_count clear_pkt = f"$OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{black_leds}*\n" try: conn.write(clear_pkt.encode('utf-8')) b["last_tx_time"] = time.time() except Exception: pass def test_led_strip(self): """Executes an energetic 4-phase test pattern across connected WS2812B LED strip: 1. Fast color wipe (Red -> Green -> Blue) 2. Knight Rider / Larson scanner comet bounce back and forth 3. GT3 Rev sweep with 3x shift flashes 4. Clean shut off to black """ def _test_worker(): connected_boards = [b for b in self.boards.values() if b.get("status") == "CONNECTED" and b.get("serial_conn")] if not connected_boards: return led_count = 300 for b in connected_boards: pins = b.get("pins", {}) led_count = max(8, min(300, int(pins.get("led_count", 300)))) break def _push_colors(color_list): hex_str = "".join(color_list[:led_count]) pkt = f"$OUT,FL:0,FR:0,BRK:0,GAS:0,LEDS:{hex_str}*\n" with self.lock: for b in self.boards.values(): if b.get("status") == "CONNECTED": conn = b.get("serial_conn") if conn and getattr(conn, 'is_open', False): try: conn.write(pkt.encode('utf-8')) b["last_tx_time"] = time.time() except Exception: pass # --- Phase 1: Fast Color Wipe (Red -> Green -> Blue) --- wipe_colors = ["FF0000", "00FF00", "0088FF"] step = max(5, led_count // 25) for c in wipe_colors: cols = ["000000"] * led_count for i in range(0, led_count + step, step): limit = min(i, led_count) for k in range(limit): cols[k] = c _push_colors(cols) time.sleep(0.012) time.sleep(0.06) # --- Phase 2: Knight Rider / Larson Scanner Comet (Forward & Back) --- comet_step = max(2, led_count // 60) for direction in [1, -1]: pos_range = range(0, led_count, comet_step) if direction == 1 else range(led_count - 1, -1, -comet_step) for center in pos_range: cols = ["000000"] * led_count for offset in range(-8, 9): idx = center + offset if 0 <= idx < led_count: dist = abs(offset) if dist == 0: cols[idx] = "FF0055" # Hot pink-red core elif dist <= 2: cols[idx] = "FF0000" # Bright red elif dist <= 5: cols[idx] = "880000" # Medium red else: cols[idx] = "220000" # Dim trail _push_colors(cols) time.sleep(0.010) # --- Phase 3: GT3 Rev Sweep with 3x Shift Strobe Flashes --- sl = self.active_config.get("shiftLights", {}) limits = self.last_engine_limits if sl.get("autoSyncCar", True) and limits and isinstance(limits, dict): max_rpm = int(limits.get("redlineRpm") or sl.get("maxRpm", 8000)) min_rpm = int(limits.get("minRpm") or (max_rpm * 0.50)) else: max_rpm = int(sl.get("maxRpm", 8000)) min_rpm = int(max_rpm * 0.50) rpm_step = max(80, int((max_rpm - min_rpm) / 40)) for r in range(min_rpm, max_rpm + rpm_step, rpm_step): self.send_test_rpm(r) time.sleep(0.022) for _ in range(3): # Electric blue / pure white shift strobe _push_colors(["00C8FF"] * led_count) time.sleep(0.07) _push_colors(["000000"] * led_count) time.sleep(0.07) # --- Phase 4: Clean shut off to black --- self.clear_leds() t = threading.Thread(target=_test_worker, daemon=True) t.start() UniversalHardwareManager = UniversalHardwareManager ESP32HardwareManager = UniversalHardwareManager ESP32_MGR = UniversalHardwareManager() def find_bossac_executable() -> Optional[str]: """Locate bossac.exe in bundled script tools, local AppData Arduino15, or PATH.""" search_paths = [ os.path.join(os.path.dirname(os.path.abspath(__file__)), "tools", "bossac.exe"), os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "public", "downloads", "tools", "bossac.exe"), os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "tools", "bossac.exe"), r"C:\_Projects\Gridpass-v5\web\public\downloads\tools\bossac.exe", ] for p in search_paths: if os.path.exists(p): return p arduino15_base = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Arduino15", "packages", "arduino", "tools", "bossac") if os.path.exists(arduino15_base): for root, _, files in os.walk(arduino15_base): if "bossac.exe" in files: return os.path.join(root, "bossac.exe") import shutil which_bossac = shutil.which("bossac.exe") or shutil.which("bossac") if which_bossac: return which_bossac # Attempt on-the-fly download of bossac.exe into %LOCALAPPDATA%\GridPass\SimOS\tools\bossac.exe target_tool_dir = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "tools") bossac_dest = os.path.join(target_tool_dir, "bossac.exe") for url in ["https://gridpass.app/downloads/tools/bossac.exe", "http://localhost:3005/downloads/tools/bossac.exe"]: try: import urllib.request req = urllib.request.Request(url, headers={'User-Agent': 'SimOS-Bridge'}) with urllib.request.urlopen(req, timeout=25) as resp: if resp.status == 200: content = resp.read() if len(content) > 100000: os.makedirs(target_tool_dir, exist_ok=True) with open(bossac_dest, "wb") as f: f.write(content) print(f"✅ [BOSSAC DOWNLOAD SUCCESS] Downloaded bossac.exe ({len(content)} bytes) to {bossac_dest}") return bossac_dest except Exception: pass return None def download_firmware_binary(bin_name: str) -> Optional[str]: """Download firmware binary on-the-fly into %LOCALAPPDATA%\GridPass\SimOS\firmware\{bin_name}.""" target_dir = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "firmware") os.makedirs(target_dir, exist_ok=True) target_path = os.path.join(target_dir, bin_name) urls = [ f"https://gridpass.app/downloads/firmware/{bin_name}", f"http://localhost:3005/downloads/firmware/{bin_name}" ] for url in urls: # Try requests first try: import requests resp = requests.get(url, headers={'User-Agent': 'SimOS-Bridge'}, timeout=20) if resp.status_code == 200 and len(resp.content) > 1000: with open(target_path, "wb") as f: f.write(resp.content) print(f"✅ [FW DOWNLOAD SUCCESS] Downloaded {bin_name} ({len(resp.content)} bytes) via requests from {url} to {target_path}") return target_path except Exception: pass # Fallback to urllib.request try: import urllib.request import ssl ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE req = urllib.request.Request(url, headers={'User-Agent': 'SimOS-Bridge'}) with urllib.request.urlopen(req, context=ctx, timeout=20) as resp: if resp.status == 200: content = resp.read() if len(content) > 1000: with open(target_path, "wb") as f: f.write(content) print(f"✅ [FW DOWNLOAD SUCCESS] Downloaded {bin_name} ({len(content)} bytes) via urllib from {url} to {target_path}") return target_path except Exception as e: print(f" (Firmware download attempt from {url} failed: {e})") return None def is_valid_com_port(device_str: Optional[str]) -> bool: """Validate that a device string is a non-empty, proper COM port (e.g. 'COM7', 'COM11').""" if not device_str or not isinstance(device_str, str): return False dev = device_str.strip().upper() return dev.startswith("COM") and len(dev) >= 4 and dev[3:].isdigit() async def trigger_samd21_bootloader(port: str) -> str: """ Triggers SAMD21 ROM bootloader via 1200 baud touch, then waits for the bootloader COM port to enumerate. Returns the bootloader COM port name (e.g. 'COM7' or newly enumerated 'COM11'). """ if not HAS_SERIAL: return port if is_valid_com_port(port) else "COM7" # Known SAMD21 bootloader USB Product IDs (Arduino MKR 1010, MKR 1000, Zero, Nano 33 IoT, etc.) # Note: Running sketch PID for MKR 1010 is 0x8054; bootloader PID is 0x0054 (0x54). BOOTLOADER_PIDS = {0x0054, 0x004E, 0x004D, 0x0057, 0x0053, 0x0055, 0x0056, 0x004F, 0x6124} # First check: Is the board ALREADY in bootloader mode? (e.g. already enumerated on COM11 or user double-tapped reset) try: for p in serial.tools.list_ports.comports(): vid = getattr(p, 'vid', None) pid = getattr(p, 'pid', None) desc = (p.description or '').lower() hwid = (p.hwid or '').lower() dev_name = (p.device or getattr(p, 'name', '') or '').strip() is_bootloader = ( (vid == 0x2341 and pid in BOOTLOADER_PIDS) or (vid == 0x03EB and pid == 0x6124) or (vid == 0x04D8) or any(k in desc or k in hwid for k in ['bossa', 'bootloader', 'sam-ba', 'samd']) ) if is_bootloader and is_valid_com_port(dev_name): print(f"🟢 [BOOTLOADER ALREADY ACTIVE] Device is already in bootloader mode at {dev_name} (VID: {hex(vid) if vid else 'None'}, PID: {hex(pid) if pid else 'None'}).") return dev_name except Exception: pass # Snapshot existing valid COM ports before 1200 baud touch initial_ports = set() try: initial_ports = {p.device.upper() for p in serial.tools.list_ports.comports() if is_valid_com_port(p.device)} except Exception: pass print(f"⚡ [BOOTLOADER TOUCH] Opening {port} at 1200 baud to trigger SAMD21 bootloader reset...") try: ser = serial.Serial() ser.port = port ser.baudrate = 1200 ser.dtr = False ser.rts = False ser.open() await asyncio.sleep(0.2) ser.close() except Exception as e: print(f" (1200 baud touch notice: {e})") # Give the USB controller time to disconnect the running sketch await asyncio.sleep(0.8) await broadcast_flash_progress(16, "preparing", f"Reset sent to {port}. Waiting for SAMD21 bootloader to re-enumerate...", port) # Wait up to 10.0 seconds for the SAMD21 bootloader COM port to fully enumerate and bind start_time = time.time() while time.time() - start_time < 10.0: await asyncio.sleep(0.3) try: current_ports = list(serial.tools.list_ports.comports()) # Check 1: Explicit bootloader signature (VID/PID or description/hwid) for p in current_ports: vid = getattr(p, 'vid', None) pid = getattr(p, 'pid', None) desc = (p.description or '').lower() hwid = (p.hwid or '').lower() dev_name = (p.device or getattr(p, 'name', '') or '').strip() is_bootloader_vid_pid = ( (vid == 0x2341 and pid in BOOTLOADER_PIDS) or (vid == 0x03EB and pid == 0x6124) or (vid == 0x04D8) ) is_bootloader_desc = any(k in desc or k in hwid for k in ['bossa', 'bootloader', 'sam-ba', 'samd']) if is_bootloader_vid_pid or is_bootloader_desc: if not is_valid_com_port(dev_name): print(f"⏳ [BOOTLOADER DETECTED] SAMD21 USB re-enumerating (VID={hex(vid) if vid else 'None'} PID={hex(pid) if pid else 'None'}), waiting for Windows COM port assignment...") continue print(f"🟢 [BOOTLOADER DETECTED] SAMD21 bootloader ready at {dev_name} (VID={hex(vid) if vid else 'None'} PID={hex(pid) if pid else 'None'}, desc='{p.description}')") await broadcast_flash_progress(25, "flashing", f"SAMD21 bootloader ready at {dev_name}. Stabilizing connection...", dev_name) await asyncio.sleep(0.8) # Allow Windows serial driver to settle return dev_name # Check 2: Any new valid COM port that was not present before 1200 baud touch new_ports = [p for p in current_ports if is_valid_com_port(p.device) and p.device.upper() not in initial_ports] if new_ports: target = new_ports[0].device.strip() print(f"🟢 [NEW COM PORT DETECTED] New serial device enumerated at {target}") await broadcast_flash_progress(25, "flashing", f"New serial port detected at {target}. Stabilizing connection...", target) await asyncio.sleep(0.8) return target except Exception: pass fallback = port if is_valid_com_port(port) else "COM7" print(f"⚠️ [BOOTLOADER WAIT TIMEOUT] Falling back to initial port {fallback}") await asyncio.sleep(0.5) return fallback async def broadcast_flash_progress(pct: int, status: str, msg: str, target_port: str): payload = json.dumps({ "type": "FLASH_PROGRESS", "status": status, "pct": pct, "message": msg, "port": target_port, "target_port": target_port, "timestamp": int(time.time() * 1000) }) legacy_payload = json.dumps({ "type": "flash_progress", "status": status, "pct": pct, "message": msg, "port": target_port, "target_port": target_port, "timestamp": int(time.time() * 1000) }) if CLIENTS: await asyncio.gather(*[c.send(payload) for c in list(CLIENTS)], return_exceptions=True) await asyncio.gather(*[c.send(legacy_payload) for c in list(CLIENTS)], return_exceptions=True) async def broadcast_flash_complete(success: bool, target_port: str, msg: str): payload = json.dumps({ "type": "FLASH_COMPLETE", "success": success, "status": "completed" if success else "error", "pct": 100 if success else 0, "message": msg, "port": target_port, "target_port": target_port, "timestamp": int(time.time() * 1000) }) legacy_payload = json.dumps({ "type": "flash_progress", "status": "completed" if success else "error", "pct": 100 if success else 0, "message": msg, "port": target_port, "target_port": target_port, "timestamp": int(time.time() * 1000) }) if CLIENTS: await asyncio.gather(*[c.send(payload) for c in list(CLIENTS)], return_exceptions=True) await asyncio.gather(*[c.send(legacy_payload) for c in list(CLIENTS)], return_exceptions=True) async def flash_board_task(port: Optional[str] = None, board_type: str = "mkr1010", firmware_version: str = "1.3.0"): """Unified automated flasher handling SAMD21 (MKR 1010 via 1200 baud touch + BOSSAC) and ESP32-S3 (esptool).""" global ESP32_MGR, CLIENTS found = ESP32_MGR.find_esp32_port() if hasattr(ESP32_MGR, 'find_esp32_port') else None target_port = port or ESP32_MGR.port or (found[0] if found else "COM7") board_type_lower = (board_type or "mkr1010").lower() print(f"\n⚡ [FLASH INITIATED] Board: {board_type_lower} on {target_port} (FW: v{firmware_version})...") # Safeguard 4: Immediate ACK packet await broadcast_flash_progress(2, "starting", "Validating COM port and tools...", target_port) await asyncio.sleep(0.05) # Safeguard 4: Verify COM port exists on host if HAS_SERIAL: try: available_ports = [p.device.upper() for p in serial.tools.list_ports.comports()] if available_ports and target_port.upper() not in available_ports: err_msg = f"COM port {target_port} not found. Verify USB cable is firmly connected." print(f"⚠️ [FLASH PRE-FLIGHT FAILED] {err_msg}") await broadcast_flash_complete(False, target_port, err_msg) ESP32_MGR.status = "ERROR" ESP32_MGR.error_message = err_msg return except Exception as pe: print(f"⚠️ [COM PORT ENUM ERROR] {pe}") ESP32_MGR.status = "FLASHING" ESP32_MGR.close_serial() await broadcast_flash_progress(5, "preparing", f"Releasing {target_port} serial connection...", target_port) await asyncio.sleep(0.4) firmware_dirs = [ os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "public", "downloads", "firmware"), os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "firmware"), r"C:\_Projects\Gridpass-v5\web\public\downloads\firmware", os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS"), os.path.join(os.path.dirname(os.path.abspath(__file__)), "firmware"), os.path.dirname(os.path.abspath(__file__)), ] flash_success = False if any(k in board_type_lower for k in ["mkr", "samd", "arduino"]): bin_names = ["SimOS_Micro_Daemon_MKR1010.bin", "SimOS_Micro_Daemon.ino.bin"] bin_path = None for d in firmware_dirs: for b_name in bin_names: p = os.path.join(d, b_name) if os.path.exists(p): bin_path = p break if bin_path: break if not bin_path: await broadcast_flash_progress(6, "preparing", f"Downloading MKR1010 firmware from cloud...", target_port) bin_path = download_firmware_binary(bin_names[0]) if not bin_path: err_msg = f"Firmware binary not found for {board_type_lower} on {target_port} (cloud download failed)." print(f"⚠️ [FLASH PRE-FLIGHT FAILED] {err_msg}") await broadcast_flash_complete(False, target_port, err_msg) ESP32_MGR.status = "ERROR" ESP32_MGR.error_message = err_msg return bossac_exe = find_bossac_executable() arduino_cli = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Arduino15", "arduino-cli.exe") has_arduino_cli = os.path.exists(arduino_cli) if not bossac_exe and not has_arduino_cli: err_msg = f"No flashing tool found (bossac.exe or arduino-cli) for SAMD21 on {target_port}." print(f"⚠️ [FLASH PRE-FLIGHT FAILED] {err_msg}") await broadcast_flash_complete(False, target_port, err_msg) ESP32_MGR.status = "ERROR" ESP32_MGR.error_message = err_msg return await broadcast_flash_progress(12, "preparing", f"Triggering 1200-baud ROM bootloader reset on {target_port}...", target_port) bootloader_port = await trigger_samd21_bootloader(target_port) if not is_valid_com_port(bootloader_port): bootloader_port = target_port if is_valid_com_port(target_port) else "COM7" await broadcast_flash_progress(26, "flashing", f"SAMD21 bootloader active on {bootloader_port}. Preparing write...", bootloader_port) creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000) if sys.platform == "win32" else 0 startupinfo = None if sys.platform == "win32": startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW startupinfo.wShowWindow = 0 # SW_HIDE if bossac_exe: try: cmd = [bossac_exe, "-i", "-d", "--port=" + bootloader_port, "-U", "true", "-i", "-e", "-w", bin_path, "-R"] print(f"[BOSSAC EXEC] {' '.join(cmd)}") await broadcast_flash_progress(28, "flashing", f"Launching bossac flasher on {bootloader_port}...", bootloader_port) proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, startupinfo=startupinfo, creationflags=creationflags ) output_buffer = "" pct = 30 import re while True: chunk = await proc.stdout.read(128) if not chunk: break decoded = chunk.decode('utf-8', errors='replace') output_buffer += decoded print(decoded, end="", flush=True) percentages = re.findall(r'(\d+)%', decoded) if percentages: raw_pct = int(percentages[-1]) if "verify" in output_buffer.lower() or "verifying" in decoded.lower(): mapped_pct = min(98, max(80, int(80 + raw_pct * 0.18))) msg = f"Verifying firmware ({raw_pct}%)..." else: mapped_pct = min(80, max(30, int(30 + raw_pct * 0.50))) msg = f"Writing firmware ({raw_pct}%)..." pct = max(pct, mapped_pct) await broadcast_flash_progress(pct, "flashing", msg, bootloader_port) elif "erase" in decoded.lower(): pct = max(pct, 35) await broadcast_flash_progress(pct, "flashing", "Erasing flash memory sectors...", bootloader_port) elif "verify" in decoded.lower(): pct = max(pct, 80) await broadcast_flash_progress(pct, "flashing", "Verifying flash contents...", bootloader_port) await proc.wait() pages_written = ("100%" in output_buffer and "pages" in output_buffer and "done in" in output_buffer) or \ ("531/531 pages" in output_buffer and "done in" in output_buffer) flash_success = (proc.returncode == 0) or pages_written if flash_success: print(f"✅ [BOSSAC SUCCESS] Firmware written successfully to {bootloader_port}.") await broadcast_flash_progress(95, "flashing", f"Firmware written. Resetting board into normal mode...", bootloader_port) await asyncio.sleep(1.5) else: err_out = output_buffer.strip()[-300:] if output_buffer.strip() else f"exit code {proc.returncode}" print(f"⚠️ [BOSSAC FAILED] exit {proc.returncode}: {err_out}") except Exception as e: print(f"⚠️ [BOSSAC EXEC ERROR] {e}") flash_success = False if not flash_success and has_arduino_cli: try: await broadcast_flash_progress(40, "flashing", f"Uploading via arduino-cli on {bootloader_port}...", bootloader_port) cmd = [arduino_cli, "upload", "-p", bootloader_port, "--fqbn", "arduino:samd:mkrwifi1010", "--input-file", bin_path] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, startupinfo=startupinfo, creationflags=creationflags ) output_buffer = "" import re while True: chunk = await proc.stdout.read(128) if not chunk: break decoded = chunk.decode('utf-8', errors='replace') output_buffer += decoded print(decoded, end="", flush=True) percentages = re.findall(r'(\d+)%', decoded) if percentages: raw_pct = int(percentages[-1]) pct = max(pct, min(95, raw_pct)) await broadcast_flash_progress(pct, "flashing", f"Writing firmware ({pct}%)...", bootloader_port) await proc.wait() flash_success = (proc.returncode == 0) except Exception as e: print(f"⚠️ [ARDUINO-CLI EXEC ERROR] {e}") if not flash_success: err_msg = f"Flashing failed for SAMD21 on {bootloader_port}. Verify board connection." print(f"⚠️ [FLASH FAILED] {err_msg}") await broadcast_flash_complete(False, target_port, err_msg) ESP32_MGR.status = "ERROR" ESP32_MGR.error_message = err_msg return else: bin_names = ["SimOS_Micro_Daemon_ESP32S3.bin", "SimOS_Micro_Daemon.ino.merged.bin", "simos_esp32s3.bin"] bin_path = None for d in firmware_dirs: for b_name in bin_names: p = os.path.join(d, b_name) if os.path.exists(p): bin_path = p break if bin_path: break if not bin_path: await broadcast_flash_progress(6, "preparing", f"Downloading ESP32-S3 firmware package from cloud...", target_port) bin_path = download_firmware_binary(bin_names[0]) if not bin_path: err_msg = f"Firmware binary not found for ESP32-S3 on {target_port} (cloud download failed)." print(f"⚠️ [FLASH PRE-FLIGHT FAILED] {err_msg}") await broadcast_flash_complete(False, target_port, err_msg) ESP32_MGR.status = "ERROR" ESP32_MGR.error_message = err_msg return if not HAS_ESPTOOL: err_msg = "esptool is not installed. Please run: pip install esptool" print(f"⚠️ [FLASH PRE-FLIGHT FAILED] {err_msg}") await broadcast_flash_complete(False, target_port, err_msg) ESP32_MGR.status = "ERROR" ESP32_MGR.error_message = err_msg return await broadcast_flash_progress(15, "flashing", f"Connecting to ESP32-S3 bootloader on {target_port}...", target_port) try: creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000) if sys.platform == "win32" else 0 startupinfo = None if sys.platform == "win32": startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW startupinfo.wShowWindow = 0 # SW_HIDE flash_offset = "0x0" cmd = [ sys.executable, "-m", "esptool", "--chip", "esp32s3", "--port", target_port, "--baud", "921600", "--before", "default_reset", "--after", "hard_reset", "write_flash", "0x0", bin_path ] print(f"[ESPTOOL EXEC] {' '.join(cmd)}") proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, startupinfo=startupinfo, creationflags=creationflags ) pct = 20 output_buffer = "" import re while True: chunk = await proc.stdout.read(128) if not chunk: break decoded = chunk.decode('utf-8', errors='replace') output_buffer += decoded percentages = re.findall(r'(\d+)%', decoded) if percentages: raw_pct = int(percentages[-1]) pct = max(pct, min(95, raw_pct)) await broadcast_flash_progress(pct, "flashing", f"Writing firmware ({pct}%)...", target_port) await proc.wait() flash_success = (proc.returncode == 0) except Exception as e: print(f"⚠️ [ESPTOOL EXEC ERROR] {e}") flash_success = False if not flash_success: err_msg = f"Flashing failed for ESP32-S3 on {target_port}. Verify board connection." print(f"⚠️ [FLASH FAILED] {err_msg}") await broadcast_flash_complete(False, target_port, err_msg) ESP32_MGR.status = "ERROR" ESP32_MGR.error_message = err_msg return # Complete flash success_msg = f"SimOS {board_type.upper()} Firmware v{firmware_version} successfully flashed on {target_port}!" await broadcast_flash_complete(True, target_port, success_msg) if 'bootloader_port' in locals() and bootloader_port and bootloader_port != target_port: await broadcast_flash_complete(True, bootloader_port, success_msg) ESP32_MGR.needs_flash = False ESP32_MGR.firmware_version = firmware_version ESP32_MGR.status = "DISCONNECTED" await asyncio.sleep(2.5) # Reconnect serial ESP32_MGR.poll_connection() status_pkt = json.dumps({ "type": "status", "source": "iracing", "bridge_version": BRIDGE_VERSION, "esp32": ESP32_MGR.get_state(), "hardware": ESP32_MGR.get_state(), "boards": ESP32_MGR.get_boards_list(), "timestamp": int(time.time() * 1000) }) if CLIENTS: await asyncio.gather(*[c.send(status_pkt) for c in list(CLIENTS)], return_exceptions=True) async def flash_esp32_firmware(firmware_version: str = "1.3.0"): await flash_board_task(port=None, board_type="esp32s3", firmware_version=firmware_version) LAST_TELEMETRY: Dict[str, Any] = {} # ============================================================================ # 4. IRACING CONFIG MANAGER, WINDOW MANAGEMENT & COCKPIT CONTROLS # ============================================================================ class IRacingConfigManager: """Dynamically locates and parses iRacing configuration files (app.ini, renderer ini, controls.cfg). Provides permanent, multi-monitor viewport awareness and accurate Drive button coordinate calculation. """ def __init__(self): self.iracing_dir = self.locate_iracing_dir() self.config_data: Dict[str, Any] = {} self.last_refresh_time = 0.0 self.refresh(force=True) def locate_iracing_dir(self) -> Optional[str]: candidates = [ os.path.join(os.environ.get("USERPROFILE", ""), "OneDrive", "Documents", "iRacing"), os.path.join(os.environ.get("USERPROFILE", ""), "Documents", "iRacing"), os.path.expanduser("~/OneDrive/Documents/iRacing"), os.path.expanduser("~/Documents/iRacing"), ] if sys.platform == "win32": try: buf = ctypes.create_unicode_buffer(260) # CSIDL_PERSONAL / CSIDL_MYDOCUMENTS = 5 ctypes.windll.shell32.SHGetFolderPathW(0, 5, 0, 0, buf) doc_path = buf.value if doc_path: candidates.append(os.path.join(doc_path, "iRacing")) except Exception: pass for c in candidates: if c and os.path.exists(os.path.join(c, "app.ini")): return os.path.normpath(c) return None def refresh(self, force: bool = False): now = time.time() if not force and (now - self.last_refresh_time < 2.0): return self.last_refresh_time = now if not self.iracing_dir or not os.path.exists(self.iracing_dir): self.iracing_dir = self.locate_iracing_dir() if not self.iracing_dir: return # 1. Active renderer INI (highest mtime among rendererDX11*.ini) rend_files = glob.glob(os.path.join(self.iracing_dir, "rendererDX11*.ini")) active_rend_file = None if rend_files: active_rend_file = max(rend_files, key=os.path.getmtime) rend_cp = configparser.ConfigParser(strict=False, comment_prefixes=(";", "#"), inline_comment_prefixes=(";", "#")) if active_rend_file and os.path.exists(active_rend_file): try: with open(active_rend_file, "r", encoding="utf-8", errors="ignore") as f: rend_cp.read_file(f) except Exception as e: print(f"[IR_CONFIG] Error reading renderer INI: {e}") # 2. Read app.ini app_cp = configparser.ConfigParser(strict=False, comment_prefixes=(";", "#"), inline_comment_prefixes=(";", "#")) app_path = os.path.join(self.iracing_dir, "app.ini") if os.path.exists(app_path): try: with open(app_path, "r", encoding="utf-8", errors="ignore") as f: app_cp.read_file(f) except Exception as e: print(f"[IR_CONFIG] Error reading app.ini: {e}") # 3. Read controls.cfg controls_path = os.path.join(self.iracing_dir, "controls.cfg") keybinds = self.parse_controls_cfg(controls_path) # Parse display values (handling both [Display] and [Graphics DX11]) fullscreen = int(rend_cp.get("Display", "fullscreen", fallback=rend_cp.get("Graphics DX11", "fullscreen", fallback="0"))) border = int(rend_cp.get("Display", "border", fallback=rend_cp.get("Graphics DX11", "border", fallback="0"))) windowed_maximized = int(rend_cp.get("Display", "windowedmaximized", fallback=rend_cp.get("Graphics DX11", "windowedmaximized", fallback="0"))) if fullscreen: display_mode = "fullscreen" width = int(rend_cp.get("Display", "fullscreenwidth", fallback=rend_cp.get("Graphics DX11", "fullscreenwidth", fallback="1920"))) height = int(rend_cp.get("Display", "fullscreenheight", fallback=rend_cp.get("Graphics DX11", "fullscreenheight", fallback="1080"))) x = 0 y = 0 elif border == 0: display_mode = "borderless" width = int(rend_cp.get("Display", "windowedwidth", fallback=rend_cp.get("Graphics DX11", "windowedwidth", fallback="1920"))) height = int(rend_cp.get("Display", "windowedheight", fallback=rend_cp.get("Graphics DX11", "windowedheight", fallback="1080"))) x = int(rend_cp.get("Display", "windowedxpos", fallback=rend_cp.get("Graphics DX11", "windowedxpos", fallback="0"))) y = int(rend_cp.get("Display", "windowedypos", fallback=rend_cp.get("Graphics DX11", "windowedypos", fallback="0"))) else: display_mode = "windowed" width = int(rend_cp.get("Display", "windowedwidth", fallback=rend_cp.get("Graphics DX11", "windowedwidth", fallback="1920"))) height = int(rend_cp.get("Display", "windowedheight", fallback=rend_cp.get("Graphics DX11", "windowedheight", fallback="1080"))) x = int(rend_cp.get("Display", "windowedxpos", fallback=rend_cp.get("Graphics DX11", "windowedxpos", fallback="0"))) y = int(rend_cp.get("Display", "windowedypos", fallback=rend_cp.get("Graphics DX11", "windowedypos", fallback="0"))) ui_scale = float(rend_cp.get("Graphics Options", "uiscale", fallback=rend_cp.get("Graphics DX11", "uiscale", fallback="100.0"))) session_ui_fullscreen = int(rend_cp.get("User Options", "sessionuifullscreen", fallback=app_cp.get("Drive Screen", "sessionuifullscreen", fallback="0"))) drive_ui_fullscreen = int(rend_cp.get("User Options", "driveuifullscreen", fallback=app_cp.get("Drive Screen", "driveuifullscreen", fallback="0"))) self.config_data = { "dir": self.iracing_dir, "active_renderer_ini": os.path.basename(active_rend_file) if active_rend_file else "N/A", "display_mode": display_mode, "windowed_maximized": bool(windowed_maximized), "viewport": {"x": x, "y": y, "width": width, "height": height}, "ui_scale": ui_scale, "session_ui_fullscreen": bool(session_ui_fullscreen), "drive_ui_fullscreen": bool(drive_ui_fullscreen), "keybinds": keybinds, "last_updated": int(now * 1000) } def parse_controls_cfg(self, path: str) -> Dict[str, Any]: keybinds = {} if not path or not os.path.exists(path): return keybinds try: with open(path, "rb") as f: data = f.read() for name in [b"Ignition", b"Starter", b"PitSpeedLimiter", b"Reset"]: pos = 0 while True: pos = data.find(name + b"\x00", pos) if pos == -1: break chunk = data[pos + len(name) + 1 : pos + len(name) + 25] if len(chunk) >= 16: input_type = int.from_bytes(chunk[8:12], "little") val = int.from_bytes(chunk[12:16], "little") if input_type == 4: # Keyboard VK keybinds[name.decode()] = {"type": "keyboard", "vk": val, "char": chr(val) if 32 <= val < 127 else ""} elif input_type == 2: # Joystick button keybinds[name.decode()] = {"type": "joystick", "btn": val} pos += len(name) + 1 except Exception as e: print(f"[IR_CONFIG] Error parsing controls.cfg: {e}") return keybinds def get_keybind_vk(self, action_name: str, default: Optional[int] = None) -> Optional[int]: kb = self.config_data.get("keybinds", {}).get(action_name) if kb and kb.get("type") == "keyboard": return kb.get("vk") return default def get_viewport_rect(self) -> Tuple[int, int, int, int]: vp = self.config_data.get("viewport", {}) return ( int(vp.get("x", 0)), int(vp.get("y", 0)), int(vp.get("width", 1920)), int(vp.get("height", 1080)) ) def get_drive_button_coords(self, client_rect: Optional[Tuple[int, int, int, int]] = None) -> Tuple[int, int]: """Calculates exact coordinates of the green DRIVE button in iRacing (top-left of interface ribbon).""" if not client_rect: client_rect = self.get_viewport_rect() x, y, w, h = client_rect ui_scale = float(self.config_data.get("ui_scale", 100.0)) scale_ratio = ui_scale / 100.0 # In triple monitor setups (w >= 3840), the UI ribbon is centered on the middle monitor if w >= 3840: monitor_w = w // 3 button_x = x + monitor_w + int(85 * scale_ratio) else: button_x = x + int(85 * scale_ratio) button_y = y + int(42 * scale_ratio) return (button_x, button_y) def get_summary(self) -> Dict[str, Any]: return dict(self.config_data) IR_CONFIG = IRacingConfigManager() # ============================================================================ # 4.1 AUTOMATED IRACING CONFIG INJECTOR (ESP32-S3 HID GAMEPAD MAPPINGS) # ============================================================================ class IRacingConfigInjector: """Automated, non-destructive configuration injector for iRacing. Inspects controls.cfg, joyCalib.yaml, and app.ini. Safely creates backups (.simos_bak) and injects standardized 12-button bindings for the GridPass ESP32-S3 USB HID Gamepad controller while strictly preserving all existing steering wheels, pedals, and shifters. """ GRIDPASS_S3_DEVICE_NAME = "GridPass S3 Controller" GRIDPASS_S3_INSTANCE_GUID = "{303A5301-0000-11EF-8001-444553540000}" GRIDPASS_S3_PRODUCT_GUID = "{1001303A-0000-0000-0000-504944564944}" MAPPINGS = [ {"action": "Reset", "btn": 1, "label": "Reset / Tow Car"}, {"action": "Ignition", "btn": 2, "label": "Ignition Toggle"}, {"action": "Starter", "btn": 3, "label": "Engine Starter"}, {"action": "PitSpeedLimiter", "btn": 4, "label": "Pit Speed Limiter"}, {"action": "TearOffVisor", "btn": 5, "label": "Visor Tear-Off"}, {"action": "AutoChat15", "btn": 6, "label": "Fast Repair Macro (#clear$ #fr$)"}, {"action": "BrakeBiasInc", "btn": 7, "label": "Brake Bias (+)"}, {"action": "BrakeBiasDec", "btn": 8, "label": "Brake Bias (-)"}, {"action": "TractionControlInc", "btn": 9, "label": "Traction Control (+)"}, {"action": "TractionControlDec", "btn": 10, "label": "Traction Control (-)"}, {"action": "ABSInc", "btn": 11, "label": "ABS (+)"}, {"action": "ABSDec", "btn": 12, "label": "ABS (-)"} ] def __init__(self, config_manager: Optional[IRacingConfigManager] = None): self.cfg_mgr = config_manager or IR_CONFIG def get_iracing_dir(self) -> Optional[str]: if hasattr(self.cfg_mgr, "iracing_dir") and self.cfg_mgr.iracing_dir and os.path.exists(self.cfg_mgr.iracing_dir): return self.cfg_mgr.iracing_dir return self.cfg_mgr.locate_iracing_dir() def guid_str_to_bytes(self, guid_str: str) -> bytes: """Converts a standard GUID string into 16 DirectInput binary bytes (little-endian).""" clean = guid_str.strip("{}") parts = clean.split("-") if len(parts) == 5: d1 = int(parts[0], 16) d2 = int(parts[1], 16) d3 = int(parts[2], 16) d4 = bytes.fromhex(parts[3] + parts[4]) return struct.pack(" str: """Converts 16 DirectInput binary bytes into a standard GUID string.""" if len(b) >= 16: d1, d2, d3 = struct.unpack(" Dict[str, bool]: """Creates .simos_bak safety copies of controls.cfg, joyCalib.yaml, and app.ini.""" ir_dir = self.get_iracing_dir() results = {"controls": False, "joycalib": False, "app_ini": False} if not ir_dir: return results import shutil for key, fname in [("controls", "controls.cfg"), ("joycalib", "joyCalib.yaml"), ("app_ini", "app.ini")]: src = os.path.join(ir_dir, fname) bak = os.path.join(ir_dir, f"{fname}.simos_bak") if os.path.exists(src): try: shutil.copy2(src, bak) results[key] = True print(f"🔒 [IR_INJECTOR] Created safety backup: {bak}") except Exception as e: print(f"⚠️ [IR_INJECTOR] Failed to backup {fname}: {e}") return results def inspect_status(self) -> Dict[str, Any]: """Inspects iRacing configuration files to check S3 controller calibration and button mapping.""" ir_dir = self.get_iracing_dir() status: Dict[str, Any] = { "iracing_found": bool(ir_dir and os.path.exists(ir_dir)), "iracing_dir": ir_dir, "joycalib_registered": False, "mapped_actions": [], "unmapped_actions": [], "backups_exist": False, "app_ini_fast_repair": False, "target_guid": self.GRIDPASS_S3_INSTANCE_GUID, "device_name": self.GRIDPASS_S3_DEVICE_NAME } if not ir_dir: return status # Check backups controls_bak = os.path.join(ir_dir, "controls.cfg.simos_bak") joy_bak = os.path.join(ir_dir, "joyCalib.yaml.simos_bak") status["backups_exist"] = os.path.exists(controls_bak) and os.path.exists(joy_bak) # Check joyCalib.yaml joy_path = os.path.join(ir_dir, "joyCalib.yaml") if os.path.exists(joy_path): try: import yaml with open(joy_path, "r", encoding="utf-8") as f: calib = yaml.safe_load(f) devs = calib.get("CalibrationInfo", {}).get("DeviceList", []) for d in devs: g = d.get("InstanceGUID", "") n = d.get("DeviceName", "") if g.upper() == self.GRIDPASS_S3_INSTANCE_GUID.upper() or "GRIDPASS" in n.upper(): status["joycalib_registered"] = True break except Exception: try: with open(joy_path, "r", encoding="utf-8", errors="ignore") as f: text = f.read() if self.GRIDPASS_S3_INSTANCE_GUID.lower() in text.lower() or "gridpass" in text.lower(): status["joycalib_registered"] = True except Exception: pass # Check app.ini app_path = os.path.join(ir_dir, "app.ini") if os.path.exists(app_path): try: with open(app_path, "r", encoding="utf-8", errors="ignore") as f: app_txt = f.read() status["app_ini_fast_repair"] = ("autoResetFastRepair=1" in app_txt) except Exception: pass # Check controls.cfg ctrl_path = os.path.join(ir_dir, "controls.cfg") if os.path.exists(ctrl_path): try: with open(ctrl_path, "rb") as f: data = f.read() target_guid_b = self.guid_str_to_bytes(self.GRIDPASS_S3_INSTANCE_GUID) for item in self.MAPPINGS: act = item["action"] btn = item["btn"] pattern = rb'(\x00|^)(' + act.encode("latin1") + rb')\x00' m = re.search(pattern, data) if m: pos = m.start(2) + len(act) + 1 if pos + 68 <= len(data): en, fl, itype = struct.unpack_from(" bool: """Ensures GridPass S3 Controller is registered in joyCalib.yaml with calibrated analog axes.""" ir_dir = self.get_iracing_dir() if not ir_dir: return False joy_path = os.path.join(ir_dir, "joyCalib.yaml") if not os.path.exists(joy_path): return False guid = guid_str or self.GRIDPASS_S3_INSTANCE_GUID s3_device = { "AxisList": [ { "Axis": 0, "AxisName": "Steering Lock (X)", "CalibCenter": 0, "CalibMax": 127, "CalibMin": -128 }, { "Axis": 1, "AxisName": "Secondary E-Brake (Y)", "CalibCenter": 0, "CalibMax": 127, "CalibMin": -128 } ], "DeviceName": self.GRIDPASS_S3_DEVICE_NAME, "InstanceGUID": guid, "ProductGUID": self.GRIDPASS_S3_PRODUCT_GUID } try: import yaml with open(joy_path, "r", encoding="utf-8") as f: calib = yaml.safe_load(f) or {} dev_list = calib.setdefault("CalibrationInfo", {}).setdefault("DeviceList", []) existing_idx = next( (i for i, d in enumerate(dev_list) if d.get("InstanceGUID", "").upper() == guid.upper()), None ) if existing_idx is not None: dev_list[existing_idx] = s3_device else: dev_list.append(s3_device) with open(joy_path, "w", encoding="utf-8") as f: yaml.dump(calib, f, default_flow_style=False, sort_keys=False) print("🎮 [IR_INJECTOR] joyCalib.yaml updated successfully.") return True except Exception as e: print(f"[IR_INJECTOR] Error updating joyCalib.yaml: {e}") return False def inject_app_ini(self) -> bool: """Configures autoResetFastRepair and AutoChatStr15 in app.ini.""" ir_dir = self.get_iracing_dir() if not ir_dir: return False app_path = os.path.join(ir_dir, "app.ini") if not os.path.exists(app_path): return False try: with open(app_path, "r", encoding="utf-8", errors="ignore") as f: content = f.read() # 1. autoResetFastRepair=1 under [Pit Service] if "autoResetFastRepair=" in content: content = re.sub(r"autoResetFastRepair=\d+", "autoResetFastRepair=1", content) elif "[Pit Service]" in content: content = content.replace("[Pit Service]", "[Pit Service]\nautoResetFastRepair=1") # 2. AutoChatStr15=#clear$ #fr$ under [Autochat Messages] if "AutoChatStr15=" in content: content = re.sub(r"AutoChatStr15=[^\r\n]*", "AutoChatStr15=#clear$ #fr$", content) elif "[Autochat Messages]" in content: content = content.replace("[Autochat Messages]", "[Autochat Messages]\nAutoChatStr15=#clear$ #fr$") with open(app_path, "w", encoding="utf-8") as f: f.write(content) print("⚙️ [IR_INJECTOR] app.ini pit service & autochat macro configured.") return True except Exception as e: print(f"[IR_INJECTOR] Error updating app.ini: {e}") return False def inject_controls_cfg(self, guid_str: Optional[str] = None) -> Dict[str, Any]: """Injects 12 DirectInput Gamepad button bindings into controls.cfg. Non-destructive: Strictly preserves all steering wheel, pedal, and shifter configurations. """ ir_dir = self.get_iracing_dir() result = {"success": False, "injected_count": 0, "actions": []} if not ir_dir: return result ctrl_path = os.path.join(ir_dir, "controls.cfg") if not os.path.exists(ctrl_path): return result guid = guid_str or self.GRIDPASS_S3_INSTANCE_GUID guid_b = self.guid_str_to_bytes(guid) try: with open(ctrl_path, "rb") as f: data = bytearray(f.read()) injected = [] for item in self.MAPPINGS: act = item["action"] btn = item["btn"] pattern = rb'(\x00|^)(' + act.encode("latin1") + rb')\x00' m = re.search(pattern, data) if not m: continue pos = m.start(2) + len(act) + 1 if pos + 68 > len(data): continue # Build 24-byte button bitmask mask = bytearray(24) byte_idx = (btn - 1) // 8 bit_val = 1 << ((btn - 1) % 8) mask[byte_idx] = bit_val # 68-byte entry: # enabled (4 bytes) = 1 # flags (4 bytes) = 6 # input_type (4 bytes) = 2 (Joystick button) # mask (24 bytes) # guid (16 bytes) # reserved (16 bytes) rec = struct.pack(" Dict[str, Any]: """Complete turnkey injection pipeline: Backups -> joyCalib -> app.ini -> controls.cfg -> reload.""" print("🚀 [IR_INJECTOR] Starting automated iRacing control injection...") backups = self.create_backups() joy_ok = self.inject_joycalib(guid_str) app_ok = self.inject_app_ini() ctrl_res = self.inject_controls_cfg(guid_str) self.cfg_mgr.refresh(force=True) return { "success": ctrl_res.get("success", False) and joy_ok, "backups": backups, "joycalib_updated": joy_ok, "app_ini_updated": app_ok, "controls_result": ctrl_res, "guid": guid_str or self.GRIDPASS_S3_INSTANCE_GUID, "timestamp": int(time.time() * 1000) } def restore_backup(self) -> Dict[str, Any]: """Restores controls.cfg, joyCalib.yaml, and app.ini from safety backups.""" ir_dir = self.get_iracing_dir() res = {"success": False, "restored": []} if not ir_dir: return res import shutil for fname in ["controls.cfg", "joyCalib.yaml", "app.ini"]: live_path = os.path.join(ir_dir, fname) bak_path = os.path.join(ir_dir, f"{fname}.simos_bak") if os.path.exists(bak_path): try: shutil.copy2(bak_path, live_path) res["restored"].append(fname) except Exception as e: print(f"⚠️ [IR_INJECTOR] Revert failed for {fname}: {e}") res["success"] = len(res["restored"]) > 0 self.cfg_mgr.refresh(force=True) return res def auto_inject_on_connect(self) -> Dict[str, Any]: """Triggered automatically when ESP32-S3 connects: verifies mappings and injects if missing.""" st = self.inspect_status() if not st.get("iracing_found"): return {"injected": False, "reason": "iRacing documents not found"} if len(st.get("unmapped_actions", [])) > 0 or not st.get("joycalib_registered"): print(f"🎮 [IR_INJECTOR] S3 Controller connected with {len(st.get('unmapped_actions', []))} unmapped actions. Auto-injecting...") res = self.inject_all() return {"injected": True, "result": res} return {"injected": False, "reason": "All 12 buttons already mapped"} IR_INJECTOR = IRacingConfigInjector() WM_IRSDK = 0 if sys.platform == "win32": try: import ctypes from ctypes import wintypes WM_IRSDK = ctypes.windll.user32.RegisterWindowMessageW("SIM_BROADCAST_MESSAGE") class WIN_POINT(ctypes.Structure): _fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)] except Exception: pass def get_mouse_pos() -> Optional[Tuple[int, int]]: if sys.platform == "win32": try: pt = WIN_POINT() ctypes.windll.user32.GetCursorPos(ctypes.byref(pt)) return (pt.x, pt.y) except Exception: pass return None def set_mouse_pos(x: int, y: int): if sys.platform == "win32": try: ctypes.windll.user32.SetCursorPos(x, y) except Exception: pass def find_iracing_window() -> Optional[int]: """Finds top-level HWND for active iRacing simulation window.""" if sys.platform != "win32": return None # 1. By class name if HAS_WIN32: try: hwnd = win32gui.FindWindow("SimWinClass", None) if hwnd and win32gui.IsWindow(hwnd): return hwnd except Exception: pass try: hwnd = ctypes.windll.user32.FindWindowW("SimWinClass", None) if hwnd: return hwnd except Exception: pass # 2. By title if HAS_WIN32: try: hwnd = win32gui.FindWindow(None, "iRacing.com Simulator") if hwnd and win32gui.IsWindow(hwnd): return hwnd except Exception: pass try: hwnd = ctypes.windll.user32.FindWindowW(None, "iRacing.com Simulator") if hwnd: return hwnd except Exception: pass # 3. Enumeration search found_hwnd = None def enum_cb(h, _): nonlocal found_hwnd if found_hwnd: return False try: title = win32gui.GetWindowText(h) if HAS_WIN32 else "" cls_name = win32gui.GetClassName(h) if HAS_WIN32 else "" if "simwinclass" in cls_name.lower() or "iracing.com simulator" in title.lower(): found_hwnd = h return False except Exception: pass return True try: if HAS_WIN32: win32gui.EnumWindows(enum_cb, 0) except Exception: pass return found_hwnd def get_iracing_client_rect(hwnd: Optional[int] = None) -> Tuple[int, int, int, int]: """Returns (screen_x, screen_y, client_width, client_height) for iRacing.""" if not hwnd: hwnd = find_iracing_window() if hwnd and sys.platform == "win32": try: if HAS_WIN32: client_rect = win32gui.GetClientRect(hwnd) # (0, 0, w, h) origin = win32gui.ClientToScreen(hwnd, (0, 0)) # (x, y) return (origin[0], origin[1], client_rect[2], client_rect[3]) else: rect = ctypes.wintypes.RECT() ctypes.windll.user32.GetClientRect(hwnd, ctypes.byref(rect)) pt = ctypes.wintypes.POINT(0, 0) ctypes.windll.user32.ClientToScreen(hwnd, ctypes.byref(pt)) return (pt.x, pt.y, rect.right - rect.left, rect.bottom - rect.top) except Exception as e: print(f"[IR_CONFIG] Error getting client rect: {e}") # Fallback to parsed iRacing configuration return IR_CONFIG.get_viewport_rect() def focus_iracing_window() -> bool: """Robustly brings iRacing simulation window to foreground, bypassing Windows foreground lock.""" if sys.platform != "win32": return False hwnd = find_iracing_window() if not hwnd: return False try: user32 = ctypes.windll.user32 curr_fg = user32.GetForegroundWindow() # Check window state is_minimized = False is_maximized = False try: if HAS_WIN32: is_minimized = bool(win32gui.IsIconic(hwnd)) is_maximized = bool(win32gui.IsZoomed(hwnd)) else: if hasattr(user32, "IsIconic"): is_minimized = bool(user32.IsIconic(hwnd)) if hasattr(user32, "IsZoomed"): is_maximized = bool(user32.IsZoomed(hwnd)) except Exception: pass # Check if iRacing is configured for windowed_maximized cfg_maximized = False try: IR_CONFIG.refresh() cfg_maximized = bool(IR_CONFIG.config_data.get("windowed_maximized", False)) except Exception: pass if curr_fg == hwnd: if cfg_maximized and not is_maximized: if HAS_WIN32: win32gui.ShowWindow(hwnd, win32con.SW_MAXIMIZE) else: user32.ShowWindow(hwnd, 3) # SW_MAXIMIZE return True # 1. Show window without un-maximizing or shrinking if is_minimized: # Only restore if it was actually minimized to taskbar if cfg_maximized: if HAS_WIN32: win32gui.ShowWindow(hwnd, win32con.SW_MAXIMIZE) else: user32.ShowWindow(hwnd, 3) # SW_MAXIMIZE else: if HAS_WIN32: win32gui.ShowWindow(hwnd, win32con.SW_RESTORE) else: user32.ShowWindow(hwnd, 9) # SW_RESTORE elif cfg_maximized and not is_maximized: # Re-maximize if window was previously shrunk/un-maximized if HAS_WIN32: win32gui.ShowWindow(hwnd, win32con.SW_MAXIMIZE) else: user32.ShowWindow(hwnd, 3) # SW_MAXIMIZE else: # NEVER shrink or un-maximize an open window! Use SW_SHOW (5) to activate without changing geometry: if HAS_WIN32: win32gui.ShowWindow(hwnd, win32con.SW_SHOW) else: user32.ShowWindow(hwnd, 5) # SW_SHOW time.sleep(0.02) # 2. Foreground lock bypass (ALT bypass via Win32 keybd_event + AttachThreadInput) bypassed = False try: # Send Alt key event to unlock SetForegroundWindow permission without COM / pythoncom side effects user32.keybd_event(0x12, 0, 0, 0) # VK_MENU down user32.keybd_event(0x12, 0, 2, 0) # VK_MENU up bypassed = True except Exception: pass if not bypassed and curr_fg: try: fg_thread = user32.GetWindowThreadProcessId(curr_fg, None) cur_thread = ctypes.windll.kernel32.GetCurrentThreadId() if fg_thread and cur_thread: user32.AttachThreadInput(cur_thread, fg_thread, True) user32.SetForegroundWindow(hwnd) user32.AttachThreadInput(cur_thread, fg_thread, False) bypassed = True except Exception: pass if not bypassed: user32.keybd_event(0x12, 0, 0, 0) # VK_MENU (ALT) user32.keybd_event(0x12, 0, 2, 0) # 3. Set foreground if HAS_WIN32: win32gui.SetForegroundWindow(hwnd) else: user32.SetForegroundWindow(hwnd) time.sleep(0.06) return user32.GetForegroundWindow() == hwnd except Exception as e: print(f"[FOCUS ERROR] {e}") return False def send_iracing_broadcast(msg_id: int, var1: int = 0, var2: int = 0): """Sends broadcast message to iRacing via Windows SIM_BROADCAST_MESSAGE.""" if sys.platform == "win32" and WM_IRSDK: try: ctypes.windll.user32.PostMessageW(0xFFFF, WM_IRSDK, msg_id, var1) except Exception as e: print(f"⚠️ [BROADCAST ERROR] {e}") def send_key_press(vk_code: int, duration: float = 0.05): """Sends key press to active or iRacing window with hardware scan code.""" if sys.platform == "win32": try: focus_iracing_window() scan = ctypes.windll.user32.MapVirtualKeyW(vk_code, 0) ctypes.windll.user32.keybd_event(vk_code, scan, 0, 0) time.sleep(duration) ctypes.windll.user32.keybd_event(vk_code, scan, 2, 0) except Exception as e: print(f"⚠️ [KEY PRESS ERROR] {e}") def execute_enter_car() -> Dict[str, Any]: """Focuses iRacing and triggers cockpit entry with dynamic coordinate calculation and key redundancy.""" IR_CONFIG.refresh() result = { "success": False, "action": "enter_car", "method": "none", "coords": None, "viewport": None, "timestamp": int(time.time() * 1000) } if sys.platform != "win32": return result try: orig_mouse = get_mouse_pos() focus_iracing_window() hwnd = find_iracing_window() client_rect = get_iracing_client_rect(hwnd) result["viewport"] = { "x": client_rect[0], "y": client_rect[1], "width": client_rect[2], "height": client_rect[3] } # Calculate exact Drive button coordinates (top ribbon, top-left) target_x, target_y = IR_CONFIG.get_drive_button_coords(client_rect) result["coords"] = {"x": target_x, "y": target_y} # 1. Click the green Drive button set_mouse_pos(target_x, target_y) time.sleep(0.05) ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0) # LEFTDOWN time.sleep(0.10) # Hold down so iRacing message loop registers the click ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0) # LEFTUP time.sleep(0.05) # In case of dual click or focus check, click once more ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0) time.sleep(0.08) ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0) # 2. Keybind redundancy: send configured Reset key if mapped to keyboard reset_vk = IR_CONFIG.get_keybind_vk("Reset") if reset_vk: time.sleep(0.05) send_key_press(reset_vk, 0.08) # 3. Shift+R hardware redundancy (standard iRacing default for Drive / Reset) time.sleep(0.05) shift_scan = ctypes.windll.user32.MapVirtualKeyW(0x10, 0) or 0x2A r_scan = ctypes.windll.user32.MapVirtualKeyW(0x52, 0) or 0x13 ctypes.windll.user32.keybd_event(0x10, shift_scan, 0, 0) # Shift down time.sleep(0.04) ctypes.windll.user32.keybd_event(0x52, r_scan, 0, 0) # R down time.sleep(0.12) ctypes.windll.user32.keybd_event(0x52, r_scan, 2, 0) # R up time.sleep(0.04) ctypes.windll.user32.keybd_event(0x10, shift_scan, 2, 0) # Shift up # 4. Restore original cursor position if orig_mouse: time.sleep(0.06) set_mouse_pos(orig_mouse[0], orig_mouse[1]) result["success"] = True result["method"] = "click+keys" print(f"🏎️ [ENTER CAR] Executed cockpit entry at ({target_x}, {target_y}) in viewport {client_rect}") except Exception as e: print(f"⚠️ [ENTER CAR ERROR] {e}") result["error"] = str(e) return result def execute_request_new_car() -> Dict[str, Any]: """Focuses iRacing and clicks Request New Car using accurate viewport coordinates.""" IR_CONFIG.refresh() result = {"success": False, "action": "request_new_car"} if sys.platform != "win32": return result try: orig_mouse = get_mouse_pos() focus_iracing_window() hwnd = find_iracing_window() client_rect = get_iracing_client_rect(hwnd) ui_scale = float(IR_CONFIG.config_data.get("ui_scale", 100.0)) scale_ratio = ui_scale / 100.0 x, y, w, h = client_rect target_x = x + (w // 2) target_y = y + int(38 * scale_ratio) set_mouse_pos(target_x, target_y) time.sleep(0.04) ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0) time.sleep(0.08) ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0) if orig_mouse: time.sleep(0.06) set_mouse_pos(orig_mouse[0], orig_mouse[1]) result["success"] = True except Exception as e: result["error"] = str(e) return result async def broadcast_smart_macro(step: int, total_steps: int, msg: str, active: bool): """Broadcasts active smart macro status packet to all WebSocket clients.""" pkt = json.dumps({ "type": "smart_macro_progress", "step": step, "totalSteps": total_steps, "total_steps": total_steps, "message": msg, "active": active, "timestamp": int(time.time() * 1000) }) if CLIENTS: await asyncio.gather(*[c.send(pkt) for c in list(CLIENTS)], return_exceptions=True) async def execute_smart_exit_macro(): """Automated vehicle shutdown and tow-to-garage macro with proper HWND focusing and hardware scan codes.""" global LAST_TELEMETRY print("🛑 [SMART EXIT] Running vehicle shutdown and pit tow macro...") if sys.platform == "win32": focus_iracing_window() # Trigger native hardware macro on ESP32-S3 Gamepad if connected try: ESP32_MGR.send_packet("$MACRO,NAME:SMART_EXIT*\n") except Exception: pass # Step 1: Cut Ignition if currently on track await broadcast_smart_macro(1, 3, "1/3 Cutting Ignition...", True) if sys.platform == "win32": ign_scan = ctypes.windll.user32.MapVirtualKeyW(0x49, 0) or 0x17 # 'I' ctypes.windll.user32.keybd_event(0x49, ign_scan, 0, 0) await asyncio.sleep(0.08) ctypes.windll.user32.keybd_event(0x49, ign_scan, 2, 0) await asyncio.sleep(0.2) # Step 2: Ensure stopped / check speed cur_speed = float(LAST_TELEMETRY.get("speedMph", 0)) if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY else 0.0 if cur_speed > 3.0: await broadcast_smart_macro(2, 3, f"2/3 Slowing Car ({cur_speed:.0f} MPH)...", True) t0 = time.time() while time.time() - t0 < 2.0: cur_speed = float(LAST_TELEMETRY.get("speedMph", 0)) if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY else 0.0 if cur_speed <= 1.0: break await asyncio.sleep(0.15) else: await broadcast_smart_macro(2, 3, "2/3 Stationary", True) await asyncio.sleep(0.1) # Step 3: Hold Escape to Tow to Garage await broadcast_smart_macro(3, 3, "3/3 Towing to Garage (Holding ESC)...", True) if sys.platform == "win32": focus_iracing_window() esc_scan = ctypes.windll.user32.MapVirtualKeyW(0x1B, 0) or 0x01 # Rapidly pump keydown events across 2.2 seconds to simulate holding the physical Escape key in iRacing t_start = time.time() while time.time() - t_start < 2.2: ctypes.windll.user32.keybd_event(0x1B, esc_scan, 0, 0) await asyncio.sleep(0.05) ctypes.windll.user32.keybd_event(0x1B, esc_scan, 2, 0) await asyncio.sleep(0.2) await broadcast_smart_macro(3, 3, "Exited to Garage ✓", False) print("🛑 [SMART EXIT] Macro completed successfully.") # ============================================================================ # 4B. STINT LIFECYCLE STATE MACHINE & IN-GAME TRANSPARENT HUD OVERLAY # ============================================================================ class StintLifecycleEngine: """Stint Lifecycle Engine & State Machine for SimOS. States: - IDLE: Rig available, awaiting driver QR scan or queue advance. - AUTHENTICATED: Driver logged in/assigned, session configured (duration, driver name, rigId). - GARAGE_WAITING: Driver in car / garage, clock has NOT started yet. - STINT_ACTIVE: Driver crossed pit exit (isOnPitRoad == False and speedMph > 25.0). Countdown timer ticks down every frame (e.g. from 15:00). - GRACE_BUFFER: Countdown reached 0:00! Driver is allowed to finish their current lap. 2-minute max safety watchdog. - TERMINATED: Triggered upon: 1. Lap completion (crossing Start/Finish line during grace buffer) 2. Spinout / severe incident (offTrack or high angular velocity > 2.5 rad/s) 3. Full stop on track (speedMph < 5.0 for > 3.0 seconds) 4. Grace buffer timeout (120 seconds elapsed) Actions: - Fire $MACRO,NAME:SMART_EXIT* (ignites tow macro) - Record stint summary (laps completed, top speed, best lap time) - Reset to IDLE after 10 seconds. """ STATE_IDLE = "IDLE" STATE_AUTHENTICATED = "AUTHENTICATED" STATE_GARAGE_WAITING = "GARAGE_WAITING" STATE_STINT_ACTIVE = "STINT_ACTIVE" STATE_GRACE_BUFFER = "GRACE_BUFFER" STATE_TERMINATED = "TERMINATED" def __init__(self): self.lock = threading.Lock() self.state: str = self.STATE_IDLE self.state_entered_time: float = time.time() # Session configuration self.rig_id: str = "1" self.driver_name: str = "Guest Driver" self.driver_handle: str = "@driver" self.driver_id: str = "" self.duration_seconds: float = 900.0 # 15 min default self.time_remaining_sec: float = 900.0 self.pricing_model: str = "per_stint" # per_stint, per_minute, per_lap self.credits_per_minute: int = 100 self.credits_per_stint: int = 1500 self.credits_per_lap: int = 150 self.default_minutes: int = 15 self.grace_buffer_enabled: bool = True self.grace_timeout_seconds: float = 120.0 self.overlay_enabled: bool = False # Runtime stint tracking self.stint_start_time: float = 0.0 self.last_timer_tick: float = time.time() self.laps_completed: int = 0 self.current_lap: int = 1 self.top_speed_mph: float = 0.0 self.best_lap_time: float = 0.0 self.last_lap_time: float = 0.0 self.initial_lap_offset: int = 0 self.initial_lap_completed_offset: int = 0 # Grace buffer & Watchdog trackers self.grace_start_time: float = 0.0 self.grace_start_lap: int = 0 self.grace_start_lap_completed: int = 0 self.grace_start_lap_dist_pct: float = 0.0 self.grace_start_incidents: int = 0 self.stationary_start_time: Optional[float] = None # Termination & History self.terminated_time: float = 0.0 self.termination_reason: str = "" self.stint_history_file = os.path.join( os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "stint_history.json" ) self.stint_config_file = os.path.join( os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "stint_config.json" ) self.stint_history: List[Dict[str, Any]] = self.load_stint_history() self.load_stint_config() def load_stint_config(self): try: if os.path.exists(self.stint_config_file): with open(self.stint_config_file, "r", encoding="utf-8") as f: cfg = json.load(f) self.rig_id = str(cfg.get("rigId", self.rig_id)) self.pricing_model = str(cfg.get("pricingModel", self.pricing_model)) self.credits_per_minute = int(cfg.get("creditsPerMinute", self.credits_per_minute)) self.credits_per_stint = int(cfg.get("creditsPerStint", self.credits_per_stint)) self.credits_per_lap = int(cfg.get("creditsPerLap", self.credits_per_lap)) self.default_minutes = int(cfg.get("defaultMinutes", self.default_minutes)) self.grace_buffer_enabled = bool(cfg.get("graceBufferEnabled", self.grace_buffer_enabled)) self.overlay_enabled = bool(cfg.get("overlayEnabled", self.overlay_enabled)) self.duration_seconds = float(self.default_minutes * 60) self.time_remaining_sec = self.duration_seconds except Exception as e: print(f"[STINT] Could not load stint config: {e}") def save_stint_config(self): try: os.makedirs(os.path.dirname(self.stint_config_file), exist_ok=True) with open(self.stint_config_file, "w", encoding="utf-8") as f: json.dump(self.get_config_dict(), f, indent=2) except Exception as e: print(f"[STINT] Could not save stint config: {e}") def load_stint_history(self) -> List[Dict[str, Any]]: try: if os.path.exists(self.stint_history_file): with open(self.stint_history_file, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return [] def save_stint_history(self): try: os.makedirs(os.path.dirname(self.stint_history_file), exist_ok=True) with open(self.stint_history_file, "w", encoding="utf-8") as f: json.dump(self.stint_history[-50:], f, indent=2) except Exception as e: print(f"[STINT] Could not save stint history: {e}") def get_config_dict(self) -> Dict[str, Any]: return { "rigId": self.rig_id, "pricingModel": self.pricing_model, "creditsPerMinute": self.credits_per_minute, "creditsPerStint": self.credits_per_stint, "creditsPerLap": self.credits_per_lap, "defaultMinutes": self.default_minutes, "graceBufferEnabled": self.grace_buffer_enabled, "graceTimeoutSeconds": self.grace_timeout_seconds, "overlayEnabled": self.overlay_enabled, } def update_config(self, cfg: Dict[str, Any]): with self.lock: if "rigId" in cfg or "rig_id" in cfg: self.rig_id = str(cfg.get("rigId") or cfg.get("rig_id")) if "pricingModel" in cfg or "pricing_model" in cfg: self.pricing_model = str(cfg.get("pricingModel") or cfg.get("pricing_model")) if "creditsPerMinute" in cfg or "credits_per_minute" in cfg: self.credits_per_minute = int(cfg.get("creditsPerMinute") or cfg.get("credits_per_minute")) if "creditsPerStint" in cfg or "credits_per_stint" in cfg: self.credits_per_stint = int(cfg.get("creditsPerStint") or cfg.get("credits_per_stint")) if "creditsPerLap" in cfg or "credits_per_lap" in cfg: self.credits_per_lap = int(cfg.get("creditsPerLap") or cfg.get("credits_per_lap")) if "defaultMinutes" in cfg or "default_minutes" in cfg: self.default_minutes = int(cfg.get("defaultMinutes") or cfg.get("default_minutes")) if self.state == self.STATE_IDLE: self.duration_seconds = float(self.default_minutes * 60) self.time_remaining_sec = self.duration_seconds if "graceBufferEnabled" in cfg or "grace_buffer_enabled" in cfg: self.grace_buffer_enabled = bool(cfg.get("graceBufferEnabled") if "graceBufferEnabled" in cfg else cfg.get("grace_buffer_enabled")) if "overlayEnabled" in cfg or "overlay_enabled" in cfg: self.overlay_enabled = bool(cfg.get("overlayEnabled") if "overlayEnabled" in cfg else cfg.get("overlay_enabled")) self.save_stint_config() print(f"[STINT] Config updated: {self.get_config_dict()}") def authenticate_driver(self, driver_name: str = "Guest Driver", driver_id: Optional[str] = None): """Authenticates a recognized driver via BLE proximity detection.""" with self.lock: self.driver_name = str(driver_name).strip() or "Guest Driver" if self.driver_name.startswith("@"): self.driver_handle = self.driver_name else: self.driver_handle = f"@{self.driver_name.lower().replace(' ', '')}" if driver_id: self.driver_id = str(driver_id) if self.state == self.STATE_IDLE: self.duration_seconds = float(self.default_minutes * 60) self.time_remaining_sec = self.duration_seconds self.laps_completed = 0 self.current_lap = 1 self.top_speed_mph = 0.0 self.best_lap_time = 0.0 self.last_lap_time = 0.0 self.stationary_start_time = None self.termination_reason = "" is_driving = False if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY: is_driving = bool(LAST_TELEMETRY.get("isOnTrack", False)) if is_driving: self.state = self.STATE_GARAGE_WAITING else: self.state = self.STATE_AUTHENTICATED self.state_entered_time = time.time() print(f"📱 [STINT AUTH] Driver {self.driver_name} ({self.driver_handle}) authenticated via BLE proximity. State -> {self.state}.") send_tray_notification( "📱 Driver Authenticated (BLE Radar)", f"{self.driver_name} seated in rig. SimOS unlocked." ) def on_driver_departed(self, driver_name: str): """Called when a recognized driver walks away from the rig prior to entering the car.""" with self.lock: if self.state == self.STATE_AUTHENTICATED: print(f"📱 [STINT] Driver {driver_name} departed prior to track entry. Resetting rig to IDLE.") self.reset_to_idle() def start_stint(self, driver_name: str = "Guest Driver", duration_minutes: float = 15.0, rig_id: Optional[str] = None, credits_per_stint: Optional[int] = None): with self.lock: self.driver_name = str(driver_name).strip() or "Guest Driver" if self.driver_name.startswith("@"): self.driver_handle = self.driver_name else: self.driver_handle = f"@{self.driver_name.lower().replace(' ', '')}" if rig_id: self.rig_id = str(rig_id) if credits_per_stint is not None: self.credits_per_stint = int(credits_per_stint) self.duration_seconds = max(60.0, float(duration_minutes) * 60.0) self.time_remaining_sec = self.duration_seconds self.laps_completed = 0 self.current_lap = 1 self.top_speed_mph = 0.0 self.best_lap_time = 0.0 self.last_lap_time = 0.0 self.stationary_start_time = None self.termination_reason = "" # Check if car is already driving or in car is_driving = False if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY: is_driving = bool(LAST_TELEMETRY.get("isOnTrack", False)) if is_driving: self.state = self.STATE_GARAGE_WAITING print(f"🏎️ [STINT] Stint AUTHENTICATED -> GARAGE_WAITING for {self.driver_name} ({self.duration_seconds/60:.0f} min). Waiting for pit exit...") else: self.state = self.STATE_AUTHENTICATED print(f"🏎️ [STINT] Stint AUTHENTICATED for {self.driver_name} ({self.duration_seconds/60:.0f} min). Waiting for driver to enter car...") self.state_entered_time = time.time() send_tray_notification( "🏎️ Stint Initialized", f"Driver: {self.driver_name} ({int(self.duration_seconds // 60)} Min) - Awaiting Pit Exit." ) def cancel_stint(self, reason: str = "Operator Cancelled"): with self.lock: if self.state != self.STATE_IDLE: self.terminate_stint(reason=reason) else: self.reset_to_idle() def terminate_stint(self, reason: str = "Stint Terminated"): """Transitions state to TERMINATED, triggers SMART_EXIT tow macro, logs summary.""" if self.state == self.STATE_TERMINATED: return self.state = self.STATE_TERMINATED self.terminated_time = time.time() self.termination_reason = reason print(f"\n🏁 [STINT TERMINATED] Reason: {reason}") print(f" Driver: {self.driver_name} | Laps: {self.laps_completed} | Top Speed: {self.top_speed_mph} MPH | Best Lap: {self.format_lap_time(self.best_lap_time)}") # 1. Fire hardware macro on ESP32-S3 try: if "ESP32_MGR" in globals() and ESP32_MGR: ESP32_MGR.send_packet("$MACRO,NAME:SMART_EXIT*\n") except Exception: pass # 2. Fire software 4-step smart exit tow macro try: global MAIN_EVENT_LOOP if MAIN_EVENT_LOOP and MAIN_EVENT_LOOP.is_running(): asyncio.run_coroutine_threadsafe(execute_smart_exit_macro(), MAIN_EVENT_LOOP) except Exception as ex: print(f"[STINT] Could not schedule smart exit macro: {ex}") # 3. Record stint summary elapsed_stint = max(0.0, self.duration_seconds - self.time_remaining_sec) summary = { "driverName": self.driver_name, "driverHandle": self.driver_handle, "rigId": self.rig_id, "lapsCompleted": self.laps_completed, "topSpeedMph": self.top_speed_mph, "bestLapTime": self.format_lap_time(self.best_lap_time), "bestLapTimeSeconds": self.best_lap_time, "durationMinutes": round(elapsed_stint / 60.0, 1), "reason": reason, "timestamp": int(time.time() * 1000), } self.stint_history.append(summary) self.save_stint_history() # 4. Toast notification send_tray_notification( "🏁 Stint Completed", f"{self.driver_name}: {self.laps_completed} Laps · Top: {self.top_speed_mph} MPH ({reason})" ) def reset_to_idle(self): self.state = self.STATE_IDLE self.driver_name = "Guest Driver" self.driver_handle = "@driver" self.driver_id = "" self.duration_seconds = float(self.default_minutes * 60) self.time_remaining_sec = self.duration_seconds self.laps_completed = 0 self.current_lap = 1 self.top_speed_mph = 0.0 self.best_lap_time = 0.0 self.last_lap_time = 0.0 self.stationary_start_time = None self.termination_reason = "" self.state_entered_time = time.time() print("[STINT] Rig reset to IDLE. In-Game Scan-to-Race QR Code online.") def evaluate_frame(self, packet: Dict[str, Any], ir_instance: Any = None): """Called every 60Hz frame to evaluate transitions, countdown, and watchdogs.""" with self.lock: now = time.time() # Handle 10-second auto-reset from TERMINATED -> IDLE if self.state == self.STATE_TERMINATED: if now - self.terminated_time >= 10.0: self.reset_to_idle() return if self.state == self.STATE_IDLE: return is_on_track = bool(packet.get("isOnTrack", False)) is_on_pit_road = bool(packet.get("isOnPitRoad", False)) speed_mph = float(packet.get("speedMph", 0.0)) lap = int(packet.get("lap", 1)) lap_completed = int(packet.get("lapCompleted", 0)) lap_dist_pct = float(packet.get("lapDistPct", 0.0)) lap_last_time = float(packet.get("lapLastLapTime", 0.0)) lap_best_time = float(packet.get("lapBestLapTime", 0.0)) yaw_rate = float(packet.get("yawRate", 0.0)) pitch_rate = float(packet.get("pitchRate", 0.0)) roll_rate = float(packet.get("rollRate", 0.0)) track_surface = int(packet.get("trackSurface", 0)) incidents = int(packet.get("incidents", 0)) # Update live metrics if speed_mph > self.top_speed_mph: self.top_speed_mph = round(speed_mph, 1) if lap_last_time > 10.0 and (self.best_lap_time <= 0.0 or lap_last_time < self.best_lap_time): self.best_lap_time = lap_last_time elif lap_best_time > 10.0 and (self.best_lap_time <= 0.0 or lap_best_time < self.best_lap_time): self.best_lap_time = lap_best_time # State: AUTHENTICATED -> GARAGE_WAITING if self.state == self.STATE_AUTHENTICATED: if is_on_track or speed_mph > 0.0: self.state = self.STATE_GARAGE_WAITING self.initial_lap_offset = lap self.initial_lap_completed_offset = lap_completed self.state_entered_time = now print(f"[STINT] Driver entered car. Transitioned to GARAGE_WAITING (awaiting pit exit blend line).") return # State: GARAGE_WAITING -> STINT_ACTIVE if self.state == self.STATE_GARAGE_WAITING: # Clock strictly starts when car crosses pit exit: isOnPitRoad == False AND speedMph > 25.0 if is_on_track and (not is_on_pit_road) and (speed_mph > 25.0): self.state = self.STATE_STINT_ACTIVE self.stint_start_time = now self.last_timer_tick = now self.initial_lap_offset = lap self.initial_lap_completed_offset = lap_completed self.stationary_start_time = None self.state_entered_time = now print(f"🚦 [STINT ACTIVE] Crossed pit exit at {speed_mph:.1f} MPH! Countdown timer started ({self.duration_seconds/60:.0f}m).") return # State: STINT_ACTIVE if self.state == self.STATE_STINT_ACTIVE: self.laps_completed = max(0, lap_completed - self.initial_lap_completed_offset) self.current_lap = max(1, lap - self.initial_lap_offset + 1) dt = now - self.last_timer_tick self.last_timer_tick = now self.time_remaining_sec = max(0.0, self.time_remaining_sec - dt) if self.time_remaining_sec <= 0.0: if self.grace_buffer_enabled: self.state = self.STATE_GRACE_BUFFER self.grace_start_time = now self.grace_start_lap = lap self.grace_start_lap_completed = lap_completed self.grace_start_lap_dist_pct = lap_dist_pct self.grace_start_incidents = incidents self.stationary_start_time = None self.state_entered_time = now print(f"⏱️ [STINT GRACE BUFFER] Timer reached 0:00! Driver finishing final lap (120s safety watchdog).") else: self.terminate_stint(reason="Stint time expired") return # State: GRACE_BUFFER (Finish Current Lap) if self.state == self.STATE_GRACE_BUFFER: self.laps_completed = max(0, lap_completed - self.initial_lap_completed_offset) self.current_lap = max(1, lap - self.initial_lap_offset + 1) # Trigger 1: Lap Completion (Crossed S/F line during grace buffer) if (lap_completed > self.grace_start_lap_completed) or (lap > self.grace_start_lap) or (self.grace_start_lap_dist_pct > 0.85 and 0.0 < lap_dist_pct < 0.15): self.terminate_stint(reason="Lap completed (Checkered Flag)") return # Trigger 2: Spinout / Severe Incident # irsdk_OffTrack = 1, high yaw angular velocity > 2.5 rad/s (~143 deg/s) if track_surface == 1 or abs(yaw_rate) > 2.5 or abs(pitch_rate) > 2.0 or abs(roll_rate) > 2.0 or incidents > self.grace_start_incidents: self.terminate_stint(reason="Spinout / off-track incident detected") return # Trigger 3: Full stop on track (speedMph < 5.0 for > 3.0 seconds) if speed_mph < 5.0: if self.stationary_start_time is None: self.stationary_start_time = now elif now - self.stationary_start_time >= 3.0: self.terminate_stint(reason="Full stop on track (>3.0s)") return else: self.stationary_start_time = None # Trigger 4: Grace buffer timeout (120 seconds max) if now - self.grace_start_time >= self.grace_timeout_seconds: self.terminate_stint(reason="Grace buffer safety watchdog timeout (120s)") return def format_time(self, seconds: float) -> str: s = max(0, int(seconds)) mins = s // 60 sec = s % 60 return f"{mins:02d}:{sec:02d}" def format_lap_time(self, seconds: float) -> str: if not seconds or seconds <= 0.0 or seconds > 3600.0: return "--:--.---" mins = int(seconds // 60) rem_sec = seconds % 60 return f"{mins}:{rem_sec:06.3f}" def get_telemetry_dict(self) -> Dict[str, Any]: with self.lock: return { "state": self.state, "timeRemainingSec": round(self.time_remaining_sec, 1), "timeRemainingFormatted": self.format_time(self.time_remaining_sec), "driverName": self.driver_name, "driverHandle": self.driver_handle, "driverId": getattr(self, "driver_id", ""), "rigId": self.rig_id, "isGraceBuffer": (self.state == self.STATE_GRACE_BUFFER), "lapsCompleted": self.laps_completed, "currentLap": self.current_lap, "topSpeedMph": self.top_speed_mph, "bestLapTime": self.format_lap_time(self.best_lap_time), "bestLapTimeSeconds": self.best_lap_time, "terminationReason": self.termination_reason if self.state == self.STATE_TERMINATED else "", "overlayEnabled": self.overlay_enabled, "pricingModel": self.pricing_model, "creditsPerStint": self.credits_per_stint, "defaultMinutes": self.default_minutes, } STINT_ENGINE = StintLifecycleEngine() class SimOSOverlay: """In-Game Transparent HUD & QR Code Overlay for SimOS. Architecture: - Runs in a detached background daemon thread with Tkinter. - Uses WS_EX_LAYERED, WS_EX_TRANSPARENT, WS_EX_NOACTIVATE, WS_EX_TOOLWINDOW so mouse clicks and input fall directly through to iRacing / steering wheels with 0 focus-stealing. - When state == 'IDLE': Renders a centered dark glass card with high-contrast QR code: "GRIDPASS SIMRIG #1", QR URL https://gridpass.app/simos?rig={rig_id}, "📲 Scan to Race · 1,500 Credits (15 Min)" - When state in ['GARAGE_WAITING', 'STINT_ACTIVE', 'GRACE_BUFFER']: Hides the big QR card and renders a narrow top-center racing HUD strip: ⏱️ 14:32 (or 🏁 FINAL LAP in pulsing gold), 👤 @driver, ⚡ 172.4 MPH, LAP 4 / BEST: 1:21.492 - When state == 'TERMINATED': Renders a finish banner with summary stats before auto-resetting to IDLE. - Toggleable via overlay_enabled flag. """ TRANSPARENT_COLOR = "#000001" def __init__(self, stint_engine: StintLifecycleEngine): self.stint_engine = stint_engine self.root: Optional[Any] = None self.canvas: Optional[Any] = None self.cached_qr_tk: Optional[Any] = None self.cached_qr_rig: Optional[str] = None self.pulse_phase: float = 0.0 self.last_topmost_refresh: float = 0.0 self.current_geometry: str = "" self.current_width: int = 1920 self.current_height: int = 1080 self.last_iracing_hwnd: Optional[int] = None self.running: bool = False def find_iracing_window(self) -> Optional[int]: """Finds the iRacing simulator window handle via SimWinClass or window title search.""" if not HAS_WIN32: return None try: # 1. Primary: official iRacing simulator window class hwnd = win32gui.FindWindow("SimWinClass", None) if hwnd and win32gui.IsWindow(hwnd) and win32gui.IsWindowVisible(hwnd): return hwnd except Exception: pass # 2. Fallback: search top-level windows containing "iRacing.com Simulator" found = None try: def _enum_win_cb(h, _): nonlocal found if found: return try: if win32gui.IsWindow(h) and win32gui.IsWindowVisible(h): txt = win32gui.GetWindowText(h) if "iRacing.com Simulator" in txt: found = h except Exception: pass win32gui.EnumWindows(_enum_win_cb, None) except Exception: pass return found def run(self): if not HAS_TKINTER: print("[OVERLAY] Tkinter is unavailable. Overlay disabled.") return try: self.root = tk.Tk() self.root.title("SimOS_Overlay") self.root.overrideredirect(True) self.root.wm_attributes("-topmost", True) try: self.root.wm_attributes("-transparentcolor", self.TRANSPARENT_COLOR) except Exception: pass self.root.configure(bg=self.TRANSPARENT_COLOR) sw = self.root.winfo_screenwidth() sh = self.root.winfo_screenheight() self.current_width = sw self.current_height = sh self.current_geometry = f"{sw}x{sh}+0+0" self.root.geometry(self.current_geometry) self.canvas = tk.Canvas( self.root, bg=self.TRANSPARENT_COLOR, highlightthickness=0, bd=0 ) self.canvas.pack(fill="both", expand=True) self.root.update_idletasks() self.apply_win32_styles() self.running = True self.tick() self.root.mainloop() except Exception as e: print(f"⚠️ [OVERLAY ERROR] Main loop terminated: {e}") def apply_win32_styles(self): if sys.platform != "win32" or not self.root: return try: wid = self.root.winfo_id() user32 = ctypes.windll.user32 parent = user32.GetParent(wid) hwnds = [wid] if parent: hwnds.append(parent) GWL_EXSTYLE = -20 WS_EX_LAYERED = 0x00080000 WS_EX_TRANSPARENT = 0x00000020 WS_EX_NOACTIVATE = 0x08000000 WS_EX_TOOLWINDOW = 0x00000080 for h in hwnds: try: cur = user32.GetWindowLongW(h, GWL_EXSTYLE) user32.SetWindowLongW(h, GWL_EXSTYLE, cur | WS_EX_LAYERED | WS_EX_TRANSPARENT | WS_EX_NOACTIVATE | WS_EX_TOOLWINDOW) except Exception: pass except Exception as e: print(f"[OVERLAY] Could not set extended window styles: {e}") def get_qr_image(self, rig_id: str) -> Optional[Any]: if not HAS_QRCODE: return None if self.cached_qr_tk and self.cached_qr_rig == rig_id: return self.cached_qr_tk try: qr_url = f"https://gridpass.app/simos?rig={rig_id}" qr = qrcode.QRCode( version=1, error_correction=qrcode.constants.ERROR_CORRECT_M, box_size=5, border=2, ) qr.add_data(qr_url) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white").convert("RGBA") self.cached_qr_tk = ImageTk.PhotoImage(img) self.cached_qr_rig = rig_id return self.cached_qr_tk except Exception as e: print(f"[OVERLAY] Failed to generate QR code image: {e}") return None def tick(self): if not self.root or not self.canvas: return try: now = time.time() self.pulse_phase = (self.pulse_phase + 0.1) % (2 * math.pi) stint = self.stint_engine.get_telemetry_dict() overlay_on = bool(stint.get("overlayEnabled", False)) if not overlay_on: self.canvas.delete("all") try: self.root.withdraw() except Exception: pass else: state = stint.get("state", "IDLE") iracing_hwnd = self.find_iracing_window() if iracing_hwnd: # iRacing is RUNNING: track window rect and size strictly try: rect = win32gui.GetWindowRect(iracing_hwnd) left, top, right, bottom = rect width = max(100, right - left) height = max(100, bottom - top) geo_str = f"{width}x{height}+{left}+{top}" if self.current_geometry != geo_str: self.root.geometry(geo_str) self.current_geometry = geo_str self.current_width = width self.current_height = height # Position directly over iRacing's window with SetWindowPos HWND_TOPMOST if now - self.last_topmost_refresh >= 1.5: self.last_topmost_refresh = now try: wid = self.root.winfo_id() ctypes.windll.user32.SetWindowPos( wid, -1, left, top, width, height, 0x0010 | 0x0040 ) except Exception: pass except Exception: pass try: self.root.deiconify() except Exception: pass cx = self.current_width // 2 self.canvas.delete("all") if state == "IDLE": self.render_idle_card(cx, self.current_height // 2, stint) elif state in ["GARAGE_WAITING", "STINT_ACTIVE", "GRACE_BUFFER"]: self.render_racing_hud(cx, stint) elif state == "TERMINATED": self.render_terminated_banner(cx, stint) else: # iRacing is NOT running if state == "IDLE": sw = self.root.winfo_screenwidth() sh = self.root.winfo_screenheight() geo_str = f"{sw}x{sh}+0+0" if self.current_geometry != geo_str: self.root.geometry(geo_str) self.current_geometry = geo_str self.current_width = sw self.current_height = sh if now - self.last_topmost_refresh >= 3.0: self.last_topmost_refresh = now try: self.root.wm_attributes("-topmost", True) except Exception: pass try: self.root.deiconify() except Exception: pass self.canvas.delete("all") self.render_idle_card(sw // 2, sh // 2, stint) else: # Window stays cleanly withdrawn when not running and not in idle standby self.canvas.delete("all") try: self.root.withdraw() except Exception: pass except Exception: pass try: self.root.after(33, self.tick) except Exception: pass def render_idle_card(self, cx: int, cy: int, stint: Dict[str, Any]): rig_id = str(stint.get("rigId", "1")) credits_val = int(stint.get("creditsPerStint", 1500)) mins_val = int(stint.get("defaultMinutes", 15)) cw = 420 ch = 480 x1 = cx - cw // 2 y1 = cy - ch // 2 x2 = cx + cw // 2 y2 = cy + ch // 2 # Card shadow & dark glass background self.canvas.create_rectangle(x1 + 4, y1 + 4, x2 + 4, y2 + 4, fill="#000000", outline="") self.canvas.create_rectangle(x1, y1, x2, y2, fill="#0d1117", outline="#1e293b", width=2) # Top racing accent stripe self.canvas.create_line(x1, y1, x2, y1, fill="#dc2626", width=4) # Header Title self.canvas.create_text( cx, y1 + 36, text=f"GRIDPASS SIMRIG #{rig_id}", fill="#f8fafc", font=("Segoe UI", 16, "bold") ) # Badge: STANDBY // SCAN TO RACE self.canvas.create_text( cx, y1 + 65, text="🟢 RIG READY · UNLOCKED", fill="#22c55e", font=("Segoe UI", 9, "bold") ) # High-contrast QR Code qr_img = self.get_qr_image(rig_id) if qr_img: # Border box behind QR for crisp contrast self.canvas.create_rectangle(cx - 105, cy - 85, cx + 105, cy + 125, fill="#ffffff", outline="#334155", width=2) self.canvas.create_image(cx, cy + 20, image=qr_img) else: # Fallback text box if QR code library isn't loaded self.canvas.create_rectangle(cx - 110, cy - 70, cx + 110, cy + 110, fill="#1e293b", outline="#dc2626", width=2) self.canvas.create_text(cx, cy + 20, text=f"https://gridpass.app/simos\n?rig={rig_id}", fill="#ffffff", font=("Segoe UI", 10, "bold"), justify="center") # Subtitle: 📲 Scan to Race · 1,500 Credits (15 Min) sub_text = f"📲 Scan to Race · {credits_val:,} Credits ({mins_val} Min)" self.canvas.create_text( cx, y2 - 58, text=sub_text, fill="#facc15", font=("Segoe UI", 12, "bold") ) # Bottom subtle hint self.canvas.create_text( cx, y2 - 28, text="Aim phone camera at QR code to start driving", fill="#94a3b8", font=("Segoe UI", 9) ) def render_racing_hud(self, cx: int, stint: Dict[str, Any]): state = stint.get("state", "STINT_ACTIVE") driver_handle = stint.get("driverHandle", "@driver") top_speed = float(stint.get("topSpeedMph", 0.0)) cur_speed = float(LAST_TELEMETRY.get("speedMph", 0.0)) if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY else top_speed display_speed = max(cur_speed, 0.0) rpm = float(LAST_TELEMETRY.get("rpm", 0.0)) if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY else 0.0 max_rpm = max(1000.0, float(LAST_TELEMETRY.get("maxRpm", 8000.0))) if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY else 8000.0 gear = int(LAST_TELEMETRY.get("gear", 0)) if "LAST_TELEMETRY" in globals() and LAST_TELEMETRY else 0 gear_str = "R" if gear == -1 else ("N" if gear == 0 else str(gear)) lap_num = stint.get("currentLap", 1) best_lap_str = stint.get("bestLapTime", "--:--.---") time_formatted = stint.get("timeRemainingFormatted", "15:00") hw = 760 hh = 50 hx1 = cx - hw // 2 hy1 = 16 hx2 = cx + hw // 2 hy2 = hy1 + hh # HUD stealth background & thin border self.canvas.create_rectangle(hx1, hy1, hx2, hy2, fill="#090d16", outline="#1e293b", width=1) # Top color stripe if state == "GRACE_BUFFER": # Pulsing gold/amber pulse_val = 0.5 + 0.5 * math.sin(self.pulse_phase * 2) pulse_color = "#fef08a" if pulse_val > 0.5 else "#eab308" self.canvas.create_line(hx1, hy1, hx2, hy1, fill=pulse_color, width=3) elif state == "GARAGE_WAITING": self.canvas.create_line(hx1, hy1, hx2, hy1, fill="#0284c7", width=2) else: self.canvas.create_line(hx1, hy1, hx2, hy1, fill="#dc2626", width=2) # 1. Timer / State Block if state == "GRACE_BUFFER": pulse_val = 0.5 + 0.5 * math.sin(self.pulse_phase * 2) t_color = "#fef08a" if pulse_val > 0.5 else "#facc15" self.canvas.create_text( hx1 + 80, hy1 + 25, text="🏁 FINAL LAP", fill=t_color, font=("Segoe UI", 12, "bold") ) elif state == "GARAGE_WAITING": self.canvas.create_text( hx1 + 80, hy1 + 25, text="⏱️ PIT READY", fill="#38bdf8", font=("Segoe UI", 12, "bold") ) else: self.canvas.create_text( hx1 + 80, hy1 + 25, text=f"⏱️ {time_formatted}", fill="#f8fafc", font=("Segoe UI", 12, "bold") ) # Divider 1 self.canvas.create_line(hx1 + 160, hy1 + 10, hx1 + 160, hy2 - 10, fill="#1e293b", width=1) # 2. Driver Badge Block self.canvas.create_text( hx1 + 240, hy1 + 25, text=f"👤 {driver_handle}", fill="#38bdf8", font=("Segoe UI", 11, "bold") ) # Divider 2 self.canvas.create_line(hx1 + 325, hy1 + 10, hx1 + 325, hy2 - 10, fill="#1e293b", width=1) # 3. Live Speed & Gear Block self.canvas.create_text( hx1 + 395, hy1 + 25, text=f"⚡ {display_speed:.0f} MPH", fill="#facc15", font=("Segoe UI", 11, "bold") ) self.canvas.create_text( hx1 + 465, hy1 + 25, text=f"G {gear_str}", fill="#e2e8f0", font=("Segoe UI", 10, "bold") ) # 4. RPM Gauge Mini Bar rpm_ratio = min(1.0, max(0.0, rpm / max_rpm)) bar_w = 60 bar_h = 8 bx1 = hx1 + 500 by1 = hy1 + 21 bx2 = bx1 + bar_w by2 = by1 + bar_h self.canvas.create_rectangle(bx1, by1, bx2, by2, fill="#1e293b", outline="#334155", width=1) fill_w = int(bar_w * rpm_ratio) if fill_w > 0: rpm_color = "#ef4444" if rpm_ratio > 0.92 else ("#eab308" if rpm_ratio > 0.80 else "#22c55e") self.canvas.create_rectangle(bx1 + 1, by1 + 1, bx1 + fill_w, by2 - 1, fill=rpm_color, outline="") # Divider 3 self.canvas.create_line(hx1 + 580, hy1 + 10, hx1 + 580, hy2 - 10, fill="#1e293b", width=1) # 5. Lap & Best Lap Block self.canvas.create_text( hx1 + 670, hy1 + 25, text=f"LAP {lap_num} · BEST: {best_lap_str}", fill="#e2e8f0", font=("Segoe UI", 10, "bold") ) def render_terminated_banner(self, cx: int, stint: Dict[str, Any]): driver_handle = stint.get("driverHandle", "@driver") laps = stint.get("lapsCompleted", 0) top_speed = float(stint.get("topSpeedMph", 0.0)) best_lap_str = stint.get("bestLapTime", "--:--.---") reason = stint.get("terminationReason", "Session Complete") hw = 700 hh = 56 hx1 = cx - hw // 2 hy1 = 16 hx2 = cx + hw // 2 hy2 = hy1 + hh self.canvas.create_rectangle(hx1, hy1, hx2, hy2, fill="#0d1117", outline="#dc2626", width=2) self.canvas.create_text( cx, hy1 + 18, text=f"🏁 STINT COMPLETE · Towing to Garage... ({reason})", fill="#f8fafc", font=("Segoe UI", 11, "bold") ) self.canvas.create_text( cx, hy1 + 38, text=f"{driver_handle} · {laps} Laps · Top: {top_speed:.1f} MPH · Best: {best_lap_str}", fill="#facc15", font=("Segoe UI", 10) ) SIMOS_OVERLAY: Optional[SimOSOverlay] = None def start_overlay_thread(): """Initializes and runs the in-game transparent HUD & Scan-to-Race QR overlay in a background daemon thread.""" global SIMOS_OVERLAY if not HAS_TKINTER: print("[OVERLAY] Tkinter is unavailable. In-Game overlay disabled.") return None def _overlay_worker(): global SIMOS_OVERLAY SIMOS_OVERLAY = SimOSOverlay(STINT_ENGINE) SIMOS_OVERLAY.run() t = threading.Thread(target=_overlay_worker, daemon=True, name="SimOS_Overlay") t.start() print("🟢 [OVERLAY] In-game transparent HUD & Scan-to-Race QR overlay active.") return t # ============================================================================ # 4.5. NATIVE WINDOWS BLUETOOTH LOW ENERGY (BLE) PROXIMITY & DRIVER RADAR # ============================================================================ class BleProximityScanner: """Native Windows Bluetooth Low Energy (BLE) Proximity Detection & Driver Radar. Architecture: - Runs continuous background BLE discovery via Bleak on Windows 10/11. - Gracefully handles absent Bluetooth adapter or uninstalled bleak. - Proximity classification: * RSSI > -55 dBm: 'seated' (In Rig / Seated) * -70 dBm <= RSSI <= -55 dBm: 'approaching' (Walking Up / Standing Near) * RSSI < -70 dBm: 'far' (Far / Leaving) - Automated Driver Authentication: * Matches against registered device tokens in %LOCALAPPDATA%\\GridPass\\SimOS\\registered_devices.json * Matches against advertised tokens starting with 'GP-PASS-' or 'GP-' * Matches manufacturer data signatures for GridPass * When a recognized driver enters proximity (RSSI > -55 dBm): - Auto-authenticates driver in STINT_ENGINE - Broadcasts 'ble_proximity_event' ('driver_detected') to Web clients and overlay * When a recognized driver departs (RSSI < -75 dBm for > 10s or unseen for > 10s): - Resets active driver state - Broadcasts 'ble_proximity_event' ('driver_departed') """ def __init__(self): self.lock = threading.Lock() self.has_adapter: bool = False self.is_scanning: bool = False self.auto_auth_enabled: bool = True self.nearby_devices: Dict[str, Dict[str, Any]] = {} self.registered_devices: List[Dict[str, Any]] = [] self.active_driver: Optional[Dict[str, Any]] = None self.last_driver_seen_time: float = 0.0 self.last_driver_rssi: int = -99 self.departure_watchdog_start: Optional[float] = None simos_dir = os.path.join(os.environ.get("LOCALAPPDATA", os.path.expanduser("~")), "GridPass", "SimOS") self.registered_devices_file = os.path.join(simos_dir, "registered_devices.json") self.load_registered_devices() def load_registered_devices(self) -> List[Dict[str, Any]]: with self.lock: if os.path.exists(self.registered_devices_file): try: with open(self.registered_devices_file, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): self.registered_devices = data elif isinstance(data, dict): self.registered_devices = data.get("devices", []) except Exception as e: print(f"[BLE] Error loading registered devices: {e}") return self.registered_devices def save_registered_device(self, device_data: Dict[str, Any]) -> Dict[str, Any]: with self.lock: token = str(device_data.get("token", "")).strip() address = str(device_data.get("address", device_data.get("mac", ""))).strip().upper() name = str(device_data.get("name", device_data.get("driverName", device_data.get("driver", "")))).strip() device_name = str(device_data.get("device", device_data.get("deviceName", ""))).strip() driver_id = str(device_data.get("id", device_data.get("driverId", ""))).strip() updated = False for dev in self.registered_devices: if (address and dev.get("address", "").upper() == address) or \ (token and dev.get("token") == token) or \ (driver_id and dev.get("id") == driver_id): if name: dev["name"] = name if address: dev["address"] = address if token: dev["token"] = token if device_name: dev["device"] = device_name dev["updatedAt"] = int(time.time() * 1000) updated = True break if not updated: new_record = { "id": driver_id or f"drv_{int(time.time())}", "name": name or "Registered Driver", "token": token or f"GP-PASS-{(name or 'DRIVER').replace(' ', '').upper()}", "address": address, "device": device_name or "Mobile Phone", "registeredAt": int(time.time() * 1000), } self.registered_devices.append(new_record) try: os.makedirs(os.path.dirname(self.registered_devices_file), exist_ok=True) with open(self.registered_devices_file, "w", encoding="utf-8") as f: json.dump({"devices": self.registered_devices}, f, indent=2) print(f"📱 [BLE] Saved registered device: {name} ({address or token})") except Exception as e: print(f"⚠️ [BLE] Failed to save registered device: {e}") return {"status": "ok", "devices": self.registered_devices} def match_driver(self, address: str, name: str, adv_data: Any) -> Optional[Dict[str, Any]]: with self.lock: for reg in self.registered_devices: reg_addr = str(reg.get("address", "")).upper() reg_token = str(reg.get("token", "")) reg_name = str(reg.get("name", "")).lower() if reg_addr and reg_addr == address: return reg if reg_token and name and reg_token.lower() in name.lower(): return reg if reg_name and name and reg_name in name.lower(): return reg # Match generic GridPass tokens: GP-PASS-* or GP-* if name: if name.startswith("GP-PASS-") or name.startswith("GP-"): token_suffix = name.replace("GP-PASS-", "").replace("GP-", "").strip() driver_name = token_suffix.replace("-", " ").replace("_", " ") or "GridPass Driver" return { "id": f"token_{token_suffix.lower()}", "name": driver_name, "token": name, "device": name, } # Check manufacturer data try: if adv_data and hasattr(adv_data, "manufacturer_data") and adv_data.manufacturer_data: for m_id, m_bytes in adv_data.manufacturer_data.items(): if b"GP" in m_bytes or b"GRIDPASS" in m_bytes: return { "id": f"beacon_{address.replace(':', '')[-6:].lower()}", "name": "GridPass Driver", "token": f"GP-{address}", "device": name or "GridPass Beacon", } except Exception: pass return None def on_advertisement(self, device: Any, adv_data: Any): address = getattr(device, "address", "").upper() if not address: return name = getattr(device, "name", None) or getattr(adv_data, "local_name", None) or "" rssi = getattr(adv_data, "rssi", -100) now = time.time() # Path loss approximation: distance = 10 ** ((-59 - rssi) / 20.0) distance_m = round(max(0.1, 10 ** ((-59.0 - float(rssi)) / 20.0)), 2) if rssi > -55: proximity = "seated" elif rssi >= -70: proximity = "approaching" else: proximity = "far" driver_match = self.match_driver(address, name, adv_data) dev_dict: Dict[str, Any] = { "address": address, "name": name or "BLE Device", "localName": getattr(adv_data, "local_name", "") or "", "rssi": rssi, "distanceMeters": distance_m, "proximity": proximity, "lastSeen": int(now * 1000), "isDriver": (driver_match is not None), } if driver_match: dev_dict["driverId"] = driver_match.get("id", "driver") dev_dict["driverName"] = driver_match.get("name", "Driver") dev_dict["driverDevice"] = driver_match.get("device", name) with self.lock: self.nearby_devices[address] = dev_dict if driver_match and self.auto_auth_enabled: self.evaluate_driver_presence(driver_match, dev_dict, rssi, proximity, now) def evaluate_driver_presence(self, driver_match: Dict[str, Any], dev_dict: Dict[str, Any], rssi: int, proximity: str, now: float): driver_id = driver_match.get("id", "driver") driver_name = driver_match.get("name", "Driver") device_name = driver_match.get("device", dev_dict.get("name", "Mobile Device")) self.last_driver_seen_time = now self.last_driver_rssi = rssi if proximity == "seated": self.departure_watchdog_start = None should_auth = False with self.lock: if self.active_driver is None or self.active_driver.get("id") != driver_id: self.active_driver = { "id": driver_id, "name": driver_name, "rssi": rssi, "proximity": proximity, "device": device_name, "authenticatedAt": int(now * 1000), } should_auth = True if should_auth: print(f"\n📱 [BLE RADAR] Driver Detected in Rig: {driver_name} (RSSI: {rssi} dBm, Proximity: {proximity})") if "STINT_ENGINE" in globals() and STINT_ENGINE: STINT_ENGINE.authenticate_driver(driver_name=driver_name, driver_id=driver_id) event_packet = { "type": "ble_proximity_event", "status": "driver_detected", "driver": { "id": driver_id, "name": driver_name, "rssi": rssi, "proximity": "seated", "device": device_name, } } self.broadcast_event(event_packet) def broadcast_event(self, packet: Dict[str, Any]): global CLIENTS, MAIN_EVENT_LOOP try: if CLIENTS and MAIN_EVENT_LOOP and MAIN_EVENT_LOOP.is_running(): payload = json.dumps(packet) coro = asyncio.gather( *[c.send(payload) for c in list(CLIENTS)], return_exceptions=True ) asyncio.run_coroutine_threadsafe(coro, MAIN_EVENT_LOOP) except Exception: pass def get_status_dict(self) -> Dict[str, Any]: with self.lock: sorted_devices = sorted( list(self.nearby_devices.values()), key=lambda d: d.get("rssi", -100), reverse=True )[:25] return { "available": HAS_BLEAK, "hasAdapter": self.has_adapter, "isScanning": self.is_scanning, "autoAuthEnabled": self.auto_auth_enabled, "activeDriver": self.active_driver, "registeredCount": len(self.registered_devices), "registeredDevices": list(self.registered_devices), "nearbyDevices": sorted_devices, "nearbyCount": len(self.nearby_devices), "timestamp": int(time.time() * 1000), } async def run_loop(self): """Continuous background BLE scanner task.""" if not HAS_BLEAK or BleakScanner is None: print("⚠️ [BLE] 'bleak' is not installed or unavailable. Run: pip install bleak") return print("📡 [BLE] Initializing Windows Bluetooth Low Energy Proximity Radar...") self.load_registered_devices() backoff = 5.0 while True: scanner = None try: self.has_adapter = True self.is_scanning = True scanner = BleakScanner(detection_callback=self.on_advertisement) await scanner.start() print("🔵 [BLE] Bluetooth Low Energy Radar active & scanning.") backoff = 5.0 while True: await asyncio.sleep(2.0) now = time.time() # Prune stale nearby devices unseen for > 20s with self.lock: stale = [addr for addr, d in self.nearby_devices.items() if (now - (d["lastSeen"] / 1000.0)) > 20.0] for addr in stale: del self.nearby_devices[addr] # Check driver departure watchdog if self.active_driver: elapsed_unseen = now - self.last_driver_seen_time is_weak_signal = (self.last_driver_rssi < -75) if is_weak_signal or elapsed_unseen >= 10.0: if self.departure_watchdog_start is None: self.departure_watchdog_start = now elif now - self.departure_watchdog_start >= 10.0 or elapsed_unseen >= 15.0: departed = self.active_driver self.active_driver = None self.departure_watchdog_start = None print(f"👋 [BLE RADAR] Driver departed from rig: {departed.get('name')} (RSSI: {self.last_driver_rssi})") if "STINT_ENGINE" in globals() and STINT_ENGINE: STINT_ENGINE.on_driver_departed(departed.get("name", "")) self.broadcast_event({ "type": "ble_proximity_event", "status": "driver_departed", "driver": { "id": departed.get("id"), "name": departed.get("name"), "rssi": self.last_driver_rssi, "proximity": "departed", "device": departed.get("device"), } }) else: self.departure_watchdog_start = None except Exception as e: if not getattr(self, "_logged_error", False): print(f"ℹ️ [BLE] Bluetooth proximity scanner idle on this host ({e})") self._logged_error = True self.is_scanning = False self.has_adapter = False if scanner: try: await scanner.stop() except Exception: pass await asyncio.sleep(60.0) BLE_SCANNER = BleProximityScanner() # ============================================================================ # 4B. BLUETOOTH LOW ENERGY HARDWARE ADVERTISER (BIDIRECTIONAL DISCOVERY) # ============================================================================ GRIDPASS_SERVICE_UUID = "0000ff01-0000-1000-8000-00805f9b34fb" class BleRigAdvertiser: """ SimOS Bidirectional Bluetooth Low Energy Hardware Node Advertiser. Enables mobile and tablet devices (Chrome on Android, Bluefy/WebBLE on iOS, and desktop browsers) to discover this PC / Sim Rig via Web Bluetooth (navigator.bluetooth.requestDevice) and check in / claim the rig in 1 tap. Advertises: - Service UUID: GRIDPASS_SERVICE_UUID ("0000ff01-0000-1000-8000-00805f9b34fb") - Company ID: 0xFF01 (GridPass Hardware Service) - Local Name: GP-Rig-{RigId} or GridPass-Rig """ def __init__(self): self.lock = threading.Lock() self.service_uuid = GRIDPASS_SERVICE_UUID self.advertised_name = self.resolve_advertised_name() self.is_advertising = False self.publisher = None self.gatt_provider = None self.error = None def get_mac_address(self) -> str: """Retrieves the host Bluetooth adapter physical MAC address via WinRT.""" if sys.platform == "win32": try: import winrt.windows.devices.bluetooth as bt adapter = bt.BluetoothAdapter.get_default_async().get() if adapter and adapter.bluetooth_address: raw_addr = adapter.bluetooth_address return ":".join(f"{(raw_addr >> (i * 8)) & 0xFF:02X}" for i in reversed(range(6))) except Exception: pass return "" def resolve_advertised_name(self) -> str: """Determines the advertised Bluetooth node name: GP-Rig-{RigId} or GridPass-Rig.""" import socket hostname = socket.gethostname() clean_host = re.sub(r'[^a-zA-Z0-9]', '', hostname)[:12] # Check for user-configured rig ID or name in local storage/settings try: simos_dir = os.path.join(os.environ.get("LOCALAPPDATA", os.path.expanduser("~")), "GridPass", "SimOS") settings_path = os.path.join(simos_dir, "simos_settings.json") if os.path.exists(settings_path): with open(settings_path, "r", encoding="utf-8") as f: cfg = json.load(f) rig_id = cfg.get("rig_id") or cfg.get("rigId") or cfg.get("id") if rig_id: clean_id = re.sub(r'[^a-zA-Z0-9-]', '', str(rig_id))[:14] return f"GP-Rig-{clean_id}" if cfg.get("rig_name"): clean_n = re.sub(r'[^a-zA-Z0-9-]', '', str(cfg["rig_name"])).strip("-")[:14] if clean_n: return f"GP-Rig-{clean_n}" except Exception: pass if clean_host: return f"GP-Rig-{clean_host}" return "GridPass-Rig" async def start(self): """Starts advertising via Windows Runtime (WinRT) GattServiceProvider and Manufacturer Data.""" with self.lock: if self.is_advertising: return print(f"📡 [BLE ADVERTISER] Initializing Bluetooth advertiser: '{self.advertised_name}' (UUID: {self.service_uuid})...") started_any = False # 1. Start WinRT GATT Service advertising with exact GRIDPASS_SERVICE_UUID try: import winrt.windows.devices.bluetooth.genericattributeprofile as gatt import uuid guid = uuid.UUID(self.service_uuid) res = await gatt.GattServiceProvider.create_async(guid) if res and res.service_provider: self.gatt_provider = res.service_provider params = gatt.GattServiceProviderAdvertisingParameters() params.is_connectable = True params.is_discoverable = True self.gatt_provider.start_advertising_with_parameters(params) started_any = True print(f"🟢 [BLE ADVERTISER] GATT Service Provider active with UUID {self.service_uuid}") except Exception as e: print(f"⚠️ [BLE ADVERTISER] GATT Service provider notice: {e}") # 2. Start WinRT Manufacturer Data Publisher with Company ID 0xFF01 + Name try: import winrt.windows.devices.bluetooth.advertisement as adv import winrt.windows.storage.streams as streams advertisement = adv.BluetoothLEAdvertisement() m_data = adv.BluetoothLEManufacturerData() m_data.company_id = 0xFF01 # 16-bit UUID 0xFF01 for GridPass writer = streams.DataWriter() name_bytes = self.advertised_name[:16].encode("utf-8") writer.write_bytes(name_bytes) m_data.data = writer.detach_buffer() advertisement.manufacturer_data.append(m_data) publisher = adv.BluetoothLEAdvertisementPublisher(advertisement) def on_status_changed(pub, args): if args.error != 0: print(f"⚠️ [BLE ADVERTISER] Status error code: {args.error}") publisher.add_status_changed(on_status_changed) publisher.start() self.publisher = publisher started_any = True print(f"🟢 [BLE ADVERTISER] Manufacturer data beacon active: '{self.advertised_name}' (Company ID: 0xFF01)") except Exception as e: print(f"⚠️ [BLE ADVERTISER] Manufacturer data publisher notice: {e}") self.error = str(e) self.is_advertising = started_any def stop(self): """Stops BLE advertising.""" with self.lock: if self.publisher: try: self.publisher.stop() except Exception: pass self.publisher = None if self.gatt_provider: try: self.gatt_provider.stop_advertising() except Exception: pass self.gatt_provider = None self.is_advertising = False print("🔴 [BLE ADVERTISER] BLE advertising stopped.") def update_name(self, new_name: str): with self.lock: clean = re.sub(r'[^a-zA-Z0-9-]', '', new_name.strip()) if clean: if not clean.startswith("GP-Rig-") and not clean.lower().startswith("gridpass"): self.advertised_name = f"GP-Rig-{clean}" else: self.advertised_name = clean if self.is_advertising: self.stop() self.start() def get_status_dict(self) -> Dict[str, Any]: with self.lock: return { "active": self.is_advertising, "name": self.advertised_name, "serviceUuid": self.service_uuid, "companyId": "0xFF01", "bluetoothMac": self.get_mac_address(), "error": self.error, } BLE_ADVERTISER = BleRigAdvertiser() # ============================================================================ # 5. WEBSOCKET SERVER & CLIENT COMMAND HANDLER # ============================================================================ async def ws_handler(websocket: Any): """Registers new WebSocket connections and listens for client commands.""" CLIENTS.add(websocket) remote_addr = getattr(websocket, 'remote_address', 'client') print(f"[WS] Client connected: {remote_addr} (Total clients: {len(CLIENTS)})") # Send initial greeting with real Windows audio output devices and ESP32 hardware state try: devs = get_audio_output_devices(force_refresh=True) init_pkt = json.dumps({ "type": "init", "source": "iracing_bridge", "audio_devices": devs, "audioDevices": devs, "esp32": ESP32_MGR.get_state(), "hardware": ESP32_MGR.get_state(), "boards": ESP32_MGR.get_boards_list(), "lighting_layers": ESP32_MGR.lighting_layers, "lighting_profiles": ESP32_MGR.load_lighting_profiles(), "machine_passport": get_machine_passport(), "machinePassport": get_machine_passport(), "iracing_config": IR_CONFIG.get_summary(), "iracingConfig": IR_CONFIG.get_summary(), "stint": STINT_ENGINE.get_telemetry_dict(), "stint_config": STINT_ENGINE.get_config_dict(), "stintConfig": STINT_ENGINE.get_config_dict(), "ble": BLE_SCANNER.get_status_dict(), "ble_advertiser": BLE_ADVERTISER.get_status_dict(), "bleAdvertiser": BLE_ADVERTISER.get_status_dict(), "timestamp": int(time.time() * 1000) }) await websocket.send(init_pkt) except Exception: pass try: async for raw_msg in websocket: try: cmd = json.loads(raw_msg) action = cmd.get("action") or cmd.get("cmd") if action in ["test_pulse", "play_effect_pulse"]: channel_id = cmd.get("channel_id", "") device_index = cmd.get("device_index") freq = float(cmd.get("freq", 45.0)) gain = float(cmd.get("gain", 0.8)) duration = float(cmd.get("duration", 1.0)) stereo = cmd.get("stereo_channel", "MONO") play_native_test_tone(device_index, freq, gain, duration, stereo) await websocket.send(json.dumps({ "type": "action_ack", "action": action, "channel_id": channel_id, "status": "playing" })) elif action == "update_tactile_config": channels = cmd.get("channels", []) m_gain = float(cmd.get("master_gain", 90.0)) muted = bool(cmd.get("muted", False)) enabled = bool(cmd.get("enabled", True)) NATIVE_ENGINE.update_config(channels, m_gain, muted, enabled) await websocket.send(json.dumps({ "type": "action_ack", "action": "update_tactile_config", "status": "applied" })) elif action == "get_audio_devices": devs = get_audio_output_devices(force_refresh=True) await websocket.send(json.dumps({ "type": "audio_devices", "audio_devices": devs, "audioDevices": devs, "timestamp": int(time.time() * 1000) })) msg_type = cmd.get("type", "") action = cmd.get("action") or cmd.get("cmd") or msg_type if action in ["flash_board", "FLASH_BOARD"] or msg_type in ["FLASH_BOARD", "flash_board"]: port = cmd.get("port") board_type = cmd.get("boardType") or cmd.get("board_type") or "mkr1010" firmware_ver = cmd.get("firmware_version") or cmd.get("firmwareVersion") or "1.3.0" asyncio.create_task(flash_board_task(port=port, board_type=board_type, firmware_version=firmware_ver)) await websocket.send(json.dumps({ "type": "action_ack", "action": "flash_board", "status": "initiated", "port": port, "boardType": board_type })) elif action in ["test_pin", "TEST_PIN"] or msg_type in ["TEST_PIN", "test_pin"]: pin = int(cmd.get("pin", 2)) duty = int(cmd.get("duty", 40)) dur = int(cmd.get("dur", cmd.get("duration", 500))) if pin == 6: ESP32_MGR.test_led_strip() else: test_pkt = f"$TEST,PIN:{pin},DUTY:{duty},DUR:{dur}*\n" ESP32_MGR.send_packet(test_pkt) await websocket.send(json.dumps({ "type": "action_ack", "action": "test_pin", "pin": pin, "duty": duty, "dur": dur, "status": "tested" })) elif action in ["test_leds", "TEST_LEDS"] or msg_type in ["TEST_LEDS", "test_leds"]: ESP32_MGR.test_led_strip() await websocket.send(json.dumps({ "type": "action_ack", "action": "test_leds", "status": "tested" })) elif action == "flash_firmware": firmware_ver = cmd.get("firmware_version", "1.3.0") asyncio.create_task(flash_board_task(port=cmd.get("port"), board_type="esp32s3", firmware_version=firmware_ver)) await websocket.send(json.dumps({ "type": "action_ack", "action": "flash_firmware", "status": "initiated" })) elif action == "stream_sim_packet": packet_str = cmd.get("packet", "") if not packet_str or packet_str in ["$CFG*", "$CFG,DEFAULT*", "$CFG,FACTORY*", "$RESET_PINS*"]: # Send default config with factory pins factory_cfg = ( f"$CFG,FAN_L={FACTORY_S3_PINS['fan_l']}," f"FAN_R={FACTORY_S3_PINS['fan_r']}," f"BRK={FACTORY_S3_PINS['brake']}," f"GAS={FACTORY_S3_PINS['gas']}," f"LED_PIN={FACTORY_S3_PINS['led_pin']}," f"LED_COUNT=300,BRIGHT=40*\n" ) ESP32_MGR.send_packet(factory_cfg) elif packet_str: if packet_str.startswith("$TEST_RPM"): rpm_val = 0 try: clean = packet_str.strip().strip("$*\r\n") for part in clean.split(","): if "=" in part: k, v = part.split("=", 1) if k.strip().upper() == "RPM": rpm_val = int(float(v.strip())) elif ":" in part: k, v = part.split(":", 1) if k.strip().upper() in ["RPM", "TEST_RPM"]: rpm_val = int(float(v.strip())) except Exception: pass ESP32_MGR.send_test_rpm(rpm_val) elif packet_str.startswith("$TEST,") and "PIN:6" in packet_str.upper(): ESP32_MGR.test_led_strip() elif packet_str.startswith("$CFG,DEFAULT") or packet_str.startswith("$CFG,FACTORY"): factory_cfg = ( f"$CFG,FAN_L={FACTORY_S3_PINS['fan_l']}," f"FAN_R={FACTORY_S3_PINS['fan_r']}," f"BRK={FACTORY_S3_PINS['brake']}," f"GAS={FACTORY_S3_PINS['gas']}," f"LED_PIN={FACTORY_S3_PINS['led_pin']}," f"LED_COUNT=300,BRIGHT=40*\n" ) ESP32_MGR.send_packet(factory_cfg) elif packet_str.startswith("$CFG,"): try: clean = packet_str.strip().strip("$*\r\n") for part in clean.split(","): if "=" in part: k, v = part.split("=", 1) k = k.strip().upper() v = v.strip() if k == "AUTO_SYNC": val = v in ["1", "true", "TRUE", "True"] ESP32_MGR.active_config["shiftLights"]["autoSyncCar"] = val elif k == "SHIFT_RPM": ESP32_MGR.active_config["shiftLights"]["shiftPointRpm"] = int(float(v)) elif k == "MAX_RPM": ESP32_MGR.active_config["shiftLights"]["maxRpm"] = int(float(v)) except Exception: pass ESP32_MGR.send_packet(packet_str) else: ESP32_MGR.send_packet(packet_str) elif action in ["align_led_segment", "ALIGN_LED_SEGMENT"]: start_led = int(cmd.get("startLed", cmd.get("start_led", 0))) end_led = int(cmd.get("endLed", cmd.get("end_led", 0))) segment = str(cmd.get("segment", "rev")) active_adjust = str(cmd.get("activeAdjust", cmd.get("active_adjust", "both"))) ESP32_MGR.render_alignment_frame(start_led, end_led, active_adjust) await websocket.send(json.dumps({ "type": "action_ack", "action": "align_led_segment", "segment": segment, "startLed": start_led, "endLed": end_led, "activeAdjust": active_adjust, "status": "aligned" })) elif action == "update_hardware_config": cfg = cmd.get("config", {}) ESP32_MGR.update_config(cfg) await websocket.send(json.dumps({ "type": "action_ack", "action": "update_hardware_config", "status": "applied", "esp32": ESP32_MGR.get_state(), "hardware": ESP32_MGR.get_state(), "boards": ESP32_MGR.get_boards_list(), })) elif action == "update_hardware_pins": pins = cmd.get("pins", {}) target_chip = cmd.get("target_chip_id") or cmd.get("chip_id") ESP32_MGR.update_hardware_pins(pins, target_chip) await websocket.send(json.dumps({ "type": "action_ack", "action": "update_hardware_pins", "status": "applied", "esp32": ESP32_MGR.get_state(), "hardware": ESP32_MGR.get_state(), "boards": ESP32_MGR.get_boards_list(), })) elif action in ["get_hardware_status", "get_hardware_pins"]: await websocket.send(json.dumps({ "type": "hardware_status", "esp32": ESP32_MGR.get_state(), "hardware": ESP32_MGR.get_state(), "boards": ESP32_MGR.get_boards_list(), })) elif action in ["update_lighting_layers", "UPDATE_LIGHTING_LAYERS"]: layers = cmd.get("layers") or cmd.get("lighting_layers") or [] ESP32_MGR.update_lighting_layers(layers) await websocket.send(json.dumps({ "type": "action_ack", "action": "update_lighting_layers", "status": "applied", "layer_count": len(ESP32_MGR.lighting_layers), "layers": ESP32_MGR.lighting_layers })) elif action in ["save_lighting_profile", "SAVE_LIGHTING_PROFILE"]: profile = cmd.get("profile") or cmd.get("data") or {} ESP32_MGR.save_lighting_profile(profile) await websocket.send(json.dumps({ "type": "action_ack", "action": "save_lighting_profile", "status": "saved", "profile_id": profile.get("id", "") if isinstance(profile, dict) else "" })) elif action in ["load_lighting_profiles", "LOAD_LIGHTING_PROFILES", "get_lighting_profiles", "GET_LIGHTING_PROFILES"]: profiles = ESP32_MGR.load_lighting_profiles() await websocket.send(json.dumps({ "type": "lighting_profiles", "profiles": profiles, "status": "loaded" })) elif action in ["get_lighting_layers", "GET_LIGHTING_LAYERS", "load_lighting_layers", "LOAD_LIGHTING_LAYERS"]: await websocket.send(json.dumps({ "type": "lighting_layers", "layers": ESP32_MGR.lighting_layers, "status": "loaded" })) elif action in ["toggle_ignition", "TOGGLE_IGNITION"]: ign_vk = IR_CONFIG.get_keybind_vk("Ignition", default=0x49) send_key_press(ign_vk) await websocket.send(json.dumps({ "type": "action_ack", "action": "toggle_ignition", "status": "sent", "vk": ign_vk })) elif action in ["toggle_pit_limiter", "TOGGLE_PIT_LIMITER"]: pit_vk = IR_CONFIG.get_keybind_vk("PitSpeedLimiter", default=0x50) send_key_press(pit_vk) if pit_vk != 0x4C: time.sleep(0.02) send_key_press(0x4C) # Compatibility await websocket.send(json.dumps({ "type": "action_ack", "action": "toggle_pit_limiter", "status": "sent", "vk": pit_vk })) elif action == "smart_exit_car": asyncio.create_task(execute_smart_exit_macro()) await websocket.send(json.dumps({ "type": "action_ack", "action": "smart_exit_car", "status": "initiated" })) elif action in ["enter_car", "ENTER_CAR"]: res = execute_enter_car() await websocket.send(json.dumps({ "type": "action_ack", "action": "enter_car", "status": "sent" if res.get("success") else "error", "details": res })) elif action in ["request_new_car", "REQUEST_NEW_CAR"]: res = execute_request_new_car() await websocket.send(json.dumps({ "type": "action_ack", "action": "request_new_car", "status": "sent" if res.get("success") else "error", "details": res })) elif action in ["get_iracing_config", "refresh_iracing_config"]: IR_CONFIG.refresh(force=True) await websocket.send(json.dumps({ "type": "iracing_config", "config": IR_CONFIG.get_summary(), "iracing_config": IR_CONFIG.get_summary(), "iracingConfig": IR_CONFIG.get_summary() })) elif action in ["inject_iracing_controls", "INJECT_IRACING_CONTROLS"]: res = IR_INJECTOR.inject_all() await websocket.send(json.dumps({ "type": "iracing_controls_injected", "status": "success" if res.get("success") else "error", "result": res })) elif action in ["get_iracing_control_status", "GET_IRACING_CONTROL_STATUS"]: res = IR_INJECTOR.inspect_status() await websocket.send(json.dumps({ "type": "iracing_control_status", "status": "ok", "details": res })) elif action in ["restore_iracing_controls_backup", "RESTORE_IRACING_CONTROLS_BACKUP"]: res = IR_INJECTOR.restore_backup() await websocket.send(json.dumps({ "type": "iracing_controls_restored", "status": "success" if res.get("success") else "error", "result": res })) elif action in ["send_joy_button", "JOY_BUTTON", "joy_button"]: btn = int(cmd.get("button") or cmd.get("btn") or 1) state = int(cmd.get("state", 1)) ESP32_MGR.send_packet(f"$JOY,BTN:{btn},STATE:{state}*\n") await websocket.send(json.dumps({ "type": "action_ack", "action": "send_joy_button", "btn": btn, "state": state })) elif action in ["run_macro", "RUN_MACRO", "hardware_macro"]: macro_name = cmd.get("name") or cmd.get("macro") or "SMART_EXIT" ESP32_MGR.send_packet(f"$MACRO,NAME:{macro_name}*\n") await websocket.send(json.dumps({ "type": "action_ack", "action": "run_macro", "name": macro_name })) elif action in ["start_stint", "START_STINT"]: dur_min = float(cmd.get("durationMinutes") or cmd.get("duration_minutes") or cmd.get("duration", 15.0)) driver = str(cmd.get("driverName") or cmd.get("driver_name") or cmd.get("driver", "Guest Driver")) rig_id = str(cmd.get("rigId") or cmd.get("rig_id") or cmd.get("rig", "1")) credits_val = int(cmd.get("creditsPerStint") or cmd.get("credits_per_stint") or cmd.get("credits", 1500)) STINT_ENGINE.start_stint( driver_name=driver, duration_minutes=dur_min, rig_id=rig_id, credits_per_stint=credits_val ) await websocket.send(json.dumps({ "type": "action_ack", "action": "start_stint", "status": "started", "stint": STINT_ENGINE.get_telemetry_dict() })) elif action in ["cancel_stint", "CANCEL_STINT", "stop_stint"]: reason = str(cmd.get("reason", "Cancelled by client")) STINT_ENGINE.cancel_stint(reason=reason) await websocket.send(json.dumps({ "type": "action_ack", "action": "cancel_stint", "status": "cancelled", "stint": STINT_ENGINE.get_telemetry_dict() })) elif action in ["update_stint_config", "UPDATE_STINT_CONFIG"]: STINT_ENGINE.update_config(cmd) await websocket.send(json.dumps({ "type": "action_ack", "action": "update_stint_config", "status": "applied", "config": STINT_ENGINE.get_config_dict(), "stint": STINT_ENGINE.get_telemetry_dict() })) elif action in ["get_stint_status", "GET_STINT_STATUS"]: await websocket.send(json.dumps({ "type": "stint_status", "status": "ok", "stint": STINT_ENGINE.get_telemetry_dict(), "config": STINT_ENGINE.get_config_dict(), "history": STINT_ENGINE.stint_history[-10:], })) elif action in ["toggle_overlay", "TOGGLE_OVERLAY", "set_overlay", "SET_OVERLAY"]: if "enabled" in cmd: STINT_ENGINE.overlay_enabled = bool(cmd["enabled"]) elif "overlayEnabled" in cmd: STINT_ENGINE.overlay_enabled = bool(cmd["overlayEnabled"]) elif "overlay_enabled" in cmd: STINT_ENGINE.overlay_enabled = bool(cmd["overlay_enabled"]) else: STINT_ENGINE.overlay_enabled = not STINT_ENGINE.overlay_enabled STINT_ENGINE.save_stint_config() await websocket.send(json.dumps({ "type": "action_ack", "action": "toggle_overlay", "status": "applied", "overlayEnabled": STINT_ENGINE.overlay_enabled, "overlay_enabled": STINT_ENGINE.overlay_enabled })) elif action in ["start_ble_scan", "START_BLE_SCAN"]: res = BLE_SCANNER.get_status_dict() await websocket.send(json.dumps({ "type": "action_ack", "action": "start_ble_scan", "status": "scanning" if res.get("isScanning") else "standby", "ble": res, })) elif action in ["register_ble_device", "REGISTER_BLE_DEVICE"]: dev_data = cmd.get("device") or cmd save_res = BLE_SCANNER.save_registered_device(dev_data) await websocket.send(json.dumps({ "type": "action_ack", "action": "register_ble_device", "status": "registered", "device": dev_data, "ble": BLE_SCANNER.get_status_dict(), })) elif action in ["get_ble_status", "GET_BLE_STATUS"]: await websocket.send(json.dumps({ "type": "ble_status", "status": "ok", "ble": BLE_SCANNER.get_status_dict(), })) elif action in ["toggle_ble_auto_auth", "TOGGLE_BLE_AUTO_AUTH"]: if "enabled" in cmd: BLE_SCANNER.auto_auth_enabled = bool(cmd["enabled"]) elif "autoAuthEnabled" in cmd: BLE_SCANNER.auto_auth_enabled = bool(cmd["autoAuthEnabled"]) else: BLE_SCANNER.auto_auth_enabled = not BLE_SCANNER.auto_auth_enabled await websocket.send(json.dumps({ "type": "action_ack", "action": "toggle_ble_auto_auth", "autoAuthEnabled": BLE_SCANNER.auto_auth_enabled, "ble": BLE_SCANNER.get_status_dict(), })) elif action in ["start_ble_advertiser", "START_BLE_ADVERTISER"]: await BLE_ADVERTISER.start() await websocket.send(json.dumps({ "type": "action_ack", "action": "start_ble_advertiser", "status": "ok", "advertiser": BLE_ADVERTISER.get_status_dict(), })) elif action in ["stop_ble_advertiser", "STOP_BLE_ADVERTISER"]: BLE_ADVERTISER.stop() await websocket.send(json.dumps({ "type": "action_ack", "action": "stop_ble_advertiser", "status": "ok", "advertiser": BLE_ADVERTISER.get_status_dict(), })) elif action in ["set_ble_rig_name", "SET_BLE_RIG_NAME"]: new_name = str(cmd.get("name", "")).strip() if new_name: BLE_ADVERTISER.update_name(new_name) await websocket.send(json.dumps({ "type": "action_ack", "action": "set_ble_rig_name", "status": "ok", "advertiser": BLE_ADVERTISER.get_status_dict(), })) elif action in ["get_ble_advertiser_status", "GET_BLE_ADVERTISER_STATUS"]: await websocket.send(json.dumps({ "type": "ble_advertiser_status", "status": "ok", "advertiser": BLE_ADVERTISER.get_status_dict(), })) except Exception as e: print(f"[WS CMD ERROR] {e}") except websockets.exceptions.ConnectionClosed: pass finally: CLIENTS.discard(websocket) print(f"[WS] Client disconnected: {remote_addr} (Remaining: {len(CLIENTS)})") async def telemetry_broadcaster(): """Main 60Hz telemetry loop acquiring iRacing memory, broadcasting JSON, and streaming audio.""" if not HAS_IRSDK: print("[ERROR] irsdk is not installed. Please install via: pip install irsdk") return ir = irsdk.IRSDK() last_ir_connected = False last_status_broadcast = 0.0 last_car_info_check = 0.0 current_car_info = { "CarPath": "", "CarScreenName": "", "CarClassShortName": "", "TrackName": "", } print("[BRIDGE] SimOS iRacing WebSocket Server active at ws://localhost:8088") print("[BRIDGE] Native Python WASAPI Audio Engine initialized.") print("[BRIDGE] Broadcasting telemetry packets at 60Hz...") last_startup_attempt = 0.0 while True: loop_start = time.time() # Continuous USB hotplug detection for hardware boards (MKR 1010, ESP32, etc.) ESP32_MGR.poll_connection() # Check iRacing connection status (throttled to 1.0s when disconnected) is_connected = False try: now = time.time() if not ir.is_connected: if now - last_startup_attempt >= 1.0: last_startup_attempt = now ir.startup() is_connected = bool(ir.is_connected) except Exception: is_connected = False if is_connected != last_ir_connected: last_ir_connected = is_connected if is_connected: print("\n🏎️ [CONNECTED] iRacing Sim Shared Memory Online! Streaming live 60Hz telemetry & native audio.") current_car_info = extract_car_session_info(ir) last_car_info_check = time.time() car_str = current_car_info.get('CarScreenName', current_car_info.get('CarPath', 'Racecar')) track_str = current_car_info.get('TrackName', 'Unknown Track') if current_car_info.get("CarPath"): print(f" Car: {car_str} [{current_car_info.get('CarClassShortName', 'Unknown')}]") print(f" Track: {track_str}") send_tray_notification( "🏎️ iRacing Session Connected", f"{car_str} @ {track_str} (60Hz telemetry live)" ) else: print("\n⏸️ [WAITING] iRacing Sim closed or in garage. Waiting for sim startup...") send_tray_notification( "⏸️ iRacing Session Ended", "Telemetry stream returned to standby." ) if is_connected and CLIENTS: try: now = time.time() if now - last_car_info_check >= 1.0 or not current_car_info.get("CarPath"): current_car_info = extract_car_session_info(ir) last_car_info_check = now ir.freeze_var_buffer_latest() packet = extract_telemetry_packet(ir, current_car_info) STINT_ENGINE.evaluate_frame(packet, ir) packet["stint"] = STINT_ENGINE.get_telemetry_dict() LAST_TELEMETRY = packet # Feed real-time packet to native Python tactile audio engine NATIVE_ENGINE.evaluate_frame(packet) # Transmit 30/60Hz packet directly to native ESP32 hardware ESP32_MGR.compute_and_send_telemetry(packet) payload = json.dumps(packet) await asyncio.gather( *[client.send(payload) for client in list(CLIENTS)], return_exceptions=True ) except Exception: pass elif not is_connected and CLIENTS: now = time.time() STINT_ENGINE.evaluate_frame({}, None) if now - last_status_broadcast >= 1.0: last_status_broadcast = now ESP32_MGR.clear_leds() devs = get_audio_output_devices() status_packet = json.dumps({ "type": "status", "source": "iracing", "bridge_version": BRIDGE_VERSION, "connected": False, "stint": STINT_ENGINE.get_telemetry_dict(), "car_info": current_car_info, "carInfo": current_car_info, "engineLimits": current_car_info.get("engineLimits") if (current_car_info and isinstance(current_car_info, dict)) else None, "engine_limits": current_car_info.get("engineLimits") if (current_car_info and isinstance(current_car_info, dict)) else None, "audio_devices": devs, "audioDevices": devs, "esp32": ESP32_MGR.get_state(), "hardware": ESP32_MGR.get_state(), "boards": ESP32_MGR.get_boards_list(), "machine_passport": get_machine_passport(), "machinePassport": get_machine_passport(), "iracing_config": IR_CONFIG.get_summary(), "iracingConfig": IR_CONFIG.get_summary(), "ble": BLE_SCANNER.get_status_dict(), "isOnTrack": False, "is_on_track": False, "message": "iRacing Sim not running (Waiting for session)", "timestamp": int(now * 1000), }) await asyncio.gather( *[client.send(status_packet) for client in list(CLIENTS)], return_exceptions=True ) else: STINT_ENGINE.evaluate_frame({}, None) # Target 60Hz execution rate (~16.66ms per frame) elapsed = time.time() - loop_start sleep_dur = max(0.001, (1.0 / 60.0) - elapsed) await asyncio.sleep(sleep_dur) def is_iracing_process_running() -> bool: """Detect if iRacing UI or Simulator process is running using native Win32 API.""" if sys.platform != "win32": return False try: import ctypes from ctypes import wintypes class PROCESSENTRY32(ctypes.Structure): _fields_ = [ ("dwSize", wintypes.DWORD), ("cntUsage", wintypes.DWORD), ("th32ProcessID", wintypes.DWORD), ("th32DefaultHeapID", ctypes.c_size_t), ("th32ModuleID", wintypes.DWORD), ("cntThreads", wintypes.DWORD), ("th32ParentProcessID", wintypes.DWORD), ("pcPriClassBase", wintypes.LONG), ("dwFlags", wintypes.DWORD), ("szExeFile", ctypes.c_char * 260) ] hSnapshot = ctypes.windll.kernel32.CreateToolhelp32Snapshot(0x00000002, 0) if hSnapshot == -1: return False entry = PROCESSENTRY32() entry.dwSize = ctypes.sizeof(PROCESSENTRY32) success = ctypes.windll.kernel32.Process32First(hSnapshot, ctypes.byref(entry)) target_exes = [b"iracingui.exe", b"iracingsim64dx11.exe", b"iracingsim64.exe"] found = False while success: exe_lower = entry.szExeFile.lower() if any(t in exe_lower for t in target_exes): found = True break success = ctypes.windll.kernel32.Process32Next(hSnapshot, ctypes.byref(entry)) ctypes.windll.kernel32.CloseHandle(hSnapshot) return found except Exception: return False async def check_and_apply_update(source_trigger: str = "periodic") -> bool: """Checks remote SimOS endpoint for a newer bridge version and auto-updates the script.""" version_endpoints = [ "http://localhost:3005/api/simos/bridge-version", "https://gridpass.app/api/simos/bridge-version", ] download_endpoints = [ "http://localhost:3005/api/simos/download-bridge", "https://gridpass.app/api/simos/download-bridge", ] remote_info = None success_idx = -1 for i, endpoint in enumerate(version_endpoints): try: req = urllib.request.Request(endpoint, headers={"User-Agent": f"SimOS-Bridge/{BRIDGE_VERSION}"}) with urllib.request.urlopen(req, timeout=4) as resp: if resp.status == 200: data = json.loads(resp.read().decode("utf-8")) remote_info = data success_idx = i break except Exception: continue if not remote_info or "version" not in remote_info: if source_trigger == "tray_menu": send_tray_notification( "⚠️ Update Server Offline", "Could not reach gridpass.app. Check your internet connection." ) return False remote_ver_str = str(remote_info["version"]) if parse_semver(remote_ver_str) > parse_semver(BRIDGE_VERSION): print(f"🔄 [AUTO-UPDATE] ({source_trigger}) Newer SimOS Bridge available: v{remote_ver_str} (Current: v{BRIDGE_VERSION})") try: dl_url = download_endpoints[success_idx] dl_req = urllib.request.Request(dl_url, headers={"User-Agent": f"SimOS-Bridge/{BRIDGE_VERSION}"}) with urllib.request.urlopen(dl_req, timeout=10) as dl_resp: if dl_resp.status == 200: new_code = dl_resp.read().decode("utf-8") compile(new_code, "", "exec") current_script_path = os.path.abspath(__file__) temp_path = current_script_path + ".tmp" with open(temp_path, "w", encoding="utf-8") as f: f.write(new_code) os.replace(temp_path, current_script_path) local_appdata = os.environ.get("LOCALAPPDATA", "") if local_appdata: installed_path = os.path.join(local_appdata, "GridPass", "SimOS", "simos_bridge.py") if os.path.abspath(installed_path) != current_script_path and os.path.exists(os.path.dirname(installed_path)): try: with open(installed_path, "w", encoding="utf-8") as f: f.write(new_code) except Exception: pass print(f"✅ [AUTO-UPDATE] Successfully updated SimOS Bridge to v{remote_ver_str}. Hot-restarting process...") send_tray_notification( "⚡ SimOS Auto-Updated", f"Successfully upgraded to v{remote_ver_str}. Running latest build." ) if TRAY_ICON: try: TRAY_ICON.stop() except Exception: pass await asyncio.sleep(1) os.execv(sys.executable, [sys.executable] + sys.argv) return True except Exception as e: print(f"⚠️ [AUTO-UPDATE ERROR] Failed to apply update: {e}") if source_trigger == "tray_menu": send_tray_notification( "⚠️ Update Failed", f"Could not apply update: {e}" ) return False else: print(f"✅ [UPDATE CHECK] ({source_trigger}) Already on latest version: v{BRIDGE_VERSION}") if source_trigger == "tray_menu": send_tray_notification( "✅ GridPass SimOS is Up to Date", f"You are running the latest version (v{BRIDGE_VERSION})." ) return False return False async def auto_updater_loop(): """Background auto-update loop: runs at startup, every 2 hours, and on iRacing/iRacingUI launch.""" await asyncio.sleep(6) try: await check_and_apply_update(source_trigger="startup") except Exception as ex: print(f"⚠️ [AUTO-UPDATE ERROR] (startup): {ex}") last_periodic_check = time.time() last_launch_check = 0.0 was_iracing_running = False while True: now = time.time() # 1. Periodic 2-hour update check if now - last_periodic_check >= 7200: last_periodic_check = now try: await check_and_apply_update(source_trigger="periodic") except Exception as ex: print(f"⚠️ [AUTO-UPDATE ERROR] (periodic): {ex}") # 2. Real-time event check: iRacingUI or Simulator launched try: is_running = is_iracing_process_running() if is_running and not was_iracing_running and (now - last_launch_check >= 60.0): last_launch_check = now print("🏎️ [LAUNCH DETECTED] iRacing started. Checking for SimOS updates...") try: await check_and_apply_update(source_trigger="iracing_launch") except Exception as ex: print(f"⚠️ [AUTO-UPDATE ERROR] (iracing_launch): {ex}") was_iracing_running = is_running except Exception: pass # Check process status every 4 seconds await asyncio.sleep(4) # ============================================================================ # 5. WINDOWS SYSTEM TRAY ICON & 1-CLICK COCKPIT LAUNCHER # ============================================================================ TRAY_ICON: Optional[Any] = None MAIN_EVENT_LOOP: Optional[asyncio.AbstractEventLoop] = None def get_cockpit_url() -> str: """Returns the primary production GridPass rig control cockpit URL.""" if os.environ.get("SIMOS_DEV_MODE") == "1": return "http://localhost:3005/simos?source=tray&installed=true" return "https://gridpass.app/simos?source=tray&installed=true" def on_tray_open_cockpit(icon=None, item=None): """Automatically opens the default browser directly to the SimOS Cockpit rig control page.""" url = get_cockpit_url() try: webbrowser.open(url) print(f"🏎️ [TRAY] Launched SimOS Cockpit in browser: {url}") except Exception as e: print(f"⚠️ [TRAY ERROR] Failed to open browser: {e}") def send_tray_notification(title: str, message: str): """Sends a native Windows toast notification with AppId 'GridPass SimOS' and emblem icon.""" if sys.platform == "win32" and HAS_WINOTIFY: try: icon_p = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "gridpass_icon_rounded.png") if not os.path.exists(icon_p): icon_p = r"C:\_Projects\Gridpass-v5\web\public\downloads\gridpass_icon_rounded.png" toast = Notification( app_id="GridPass SimOS", title=title, msg=message, icon=icon_p if os.path.exists(icon_p) else "" ) toast.show() return except Exception: pass global TRAY_ICON if TRAY_ICON: try: TRAY_ICON.notify(title, message) except Exception: pass def on_tray_check_updates(icon=None, item=None): """Triggers an immediate update check against remote endpoints.""" print("🔄 [TRAY] Manual update check requested from system tray.") global MAIN_EVENT_LOOP if MAIN_EVENT_LOOP and MAIN_EVENT_LOOP.is_running(): asyncio.run_coroutine_threadsafe(check_and_apply_update(source_trigger="tray_menu"), MAIN_EVENT_LOOP) else: threading.Thread(target=lambda: asyncio.run(check_and_apply_update(source_trigger="tray_menu")), daemon=True).start() def on_tray_exit(icon=None, item=None): """Clean shutdown of SimOS Bridge from system tray.""" print("\n[SHUTDOWN] Exiting SimOS Bridge via System Tray...") global TRAY_ICON, MAIN_EVENT_LOOP if icon: try: icon.stop() except Exception: pass elif TRAY_ICON: try: TRAY_ICON.stop() except Exception: pass if MAIN_EVENT_LOOP and MAIN_EVENT_LOOP.is_running(): try: MAIN_EVENT_LOOP.call_soon_threadsafe(MAIN_EVENT_LOOP.stop) except Exception: pass time.sleep(0.15) os._exit(0) def get_tray_status_label(item=None) -> str: """Returns dynamic status label for system tray menu.""" if HAS_SOUNDDEVICE: return "🟢 Status: 60Hz WASAPI Online" return "🟡 Status: 60Hz Audio Fallback" def create_tray_icon_image(size: int = 64) -> Any: """Generates a crisp, vibrant, high-contrast system tray icon: Bright circular crimson red background (#dc2626) with a crisp white lightning bolt in the center and a high-contrast white border with a subtle dark halo. Guarantees 100% visibility on both Windows 11 Dark Mode and Light Mode taskbars. """ scale = 4 s = max(128, size * scale) img = Image.new("RGBA", (s, s), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) pad = max(2, int(s * 0.018)) outer_box = [pad, pad, s - pad, s - pad] # Outer subtle dark outline for 100% crispness & separation on light backgrounds draw.ellipse(outer_box, fill=(255, 255, 255, 255), outline=(15, 23, 42, 220), width=max(2, int(s * 0.02))) # Inner bright crimson red background (#dc2626) border_w = max(4, int(s * 0.045)) inner_box = [pad + border_w, pad + border_w, s - pad - border_w, s - pad - border_w] crimson = (220, 38, 38, 255) # #dc2626 draw.ellipse(inner_box, fill=crimson) # Crisp white lightning bolt in center white = (255, 255, 255, 255) bolt = [ (s * 0.55, s * 0.20), # Top tip (s * 0.35, s * 0.52), # Mid-left outer (s * 0.49, s * 0.52), # Mid jog right (s * 0.43, s * 0.80), # Bottom tip (s * 0.65, s * 0.47), # Mid-right outer (s * 0.52, s * 0.47), # Mid jog left ] draw.polygon(bolt, fill=white) # High-quality Lanczos downsampling for razor-sharp anti-aliased edges final_icon = img.resize((size, size), Image.Resampling.LANCZOS) # Also persist to disk so Windows toast notifications (winotify) can display this emblem try: save_paths = [ os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "gridpass_icon_rounded.png"), os.path.join(os.path.dirname(os.path.abspath(__file__)), "gridpass_icon_rounded.png"), r"C:\_Projects\Gridpass-v5\web\public\downloads\gridpass_icon_rounded.png", ] for sp in save_paths: try: os.makedirs(os.path.dirname(sp), exist_ok=True) final_icon.save(sp, format="PNG") except Exception: pass except Exception: pass return final_icon def on_tray_launch_overlay(icon=None, item=None): """Launches the native transparent in-game overlay over iRacing.""" try: candidates = [ os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "simos_overlay_window.py"), os.path.join(os.path.dirname(os.path.abspath(__file__)), "simos_overlay_window.py"), r"C:\_Projects\Gridpass-v5\web\scripts\simos_overlay_window.py", ] script_p = None for c in candidates: if os.path.exists(c): script_p = c break if script_p: subprocess.Popen([sys.executable, script_p, "--mode", "hud"]) print(f"🚀 [TRAY] Launched Native In-Game Overlay: {script_p}") else: webbrowser.open("http://localhost:3005/simos/overlay?mode=hud") except Exception as e: print(f"⚠️ [TRAY ERROR] Failed to launch overlay: {e}") def on_tray_view_log(icon=None, item=None): """Opens the active bridge output log in Notepad.""" log_p = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS", "bridge_output.log") if os.path.exists(log_p): try: os.startfile(log_p) return except Exception: pass webbrowser.open("http://localhost:3005/simos") def setup_system_tray(): """Initializes and runs the Windows system tray icon in a non-blocking background daemon thread.""" global TRAY_ICON if not HAS_TRAY: print("[TRAY] System tray support disabled (missing pystray or Pillow).") return None try: icon_img = create_tray_icon_image() menu = Menu( item('🏁 Launch In-Game HUD Overlay', on_tray_launch_overlay, default=True), item('🏎️ Open Rig Control Cockpit', on_tray_open_cockpit), item('📊 View Telemetry Log / Console', on_tray_view_log), Menu.SEPARATOR, item('🌐 GridPass (gridpass.app)', lambda icon, item: webbrowser.open(get_cockpit_url())), item(get_tray_status_label, None, enabled=False), item('🔄 Check for Updates', on_tray_check_updates), Menu.SEPARATOR, item('❌ Exit SimOS Bridge', on_tray_exit), ) tray_icon = pystray.Icon( name="GridPass_SimOS", icon=icon_img, title="GridPass // SimOS Hardware Engine (gridpass.app)", menu=menu ) tray_icon.run_detached() TRAY_ICON = tray_icon print("🟢 [SYSTEM TRAY] GridPass SimOS Tray Icon registered in Windows Notification Area.") # Fire native Windows notification toast to alert user where the icon is def notify_user(): time.sleep(1.0) try: send_tray_notification( "🏎️ SimOS Companion Active", "Rig Control companion is running. Look for the bright red GridPass icon in your system tray." ) except Exception: try: tray_icon.notify( "🏎️ SimOS Companion Active", "Rig Control companion is running. Look for the bright red GridPass icon in your system tray." ) except Exception: pass threading.Thread(target=notify_user, daemon=True).start() return tray_icon except Exception as e: print(f"⚠️ [SYSTEM TRAY ERROR] Could not initialize tray icon: {e}") return None async def recover_ws_port_loop(server_holder: List[Any]): """Retries binding to port 8088 if port was occupied on startup.""" while server_holder[0] is None: await asyncio.sleep(4) try: srv = await websockets.serve(ws_handler, "0.0.0.0", 8088) server_holder[0] = srv print("🟢 [PORT 8088 RECOVERED] Successfully bound WebSocket server on ws://0.0.0.0:8088!") send_tray_notification( "🟢 SimOS Server Online", "Successfully bound to ws://0.0.0.0:8088." ) await srv.wait_closed() break except OSError: pass async def main(): global MAIN_EVENT_LOOP MAIN_EVENT_LOOP = asyncio.get_running_loop() def asyncio_err_handler(loop, context): msg = context.get("exception", context.get("message", "Unknown asyncio error")) print(f"⚠️ [ASYNCIO ERROR] {msg}") MAIN_EVENT_LOOP.set_exception_handler(asyncio_err_handler) # Launch Windows Taskbar System Tray Icon (non-blocking detached thread) setup_system_tray() # In-Game overlay is disabled by default on startup to prevent rogue black monitor windows. # Users view the broadcast overlay via /simos/overlay or explicit toggle. # start_overlay_thread() # Launch Native Windows BLE Hardware Node Advertiser (GridPass Service UUID: 0000ff01-0000-1000-8000-00805f9b34fb) try: await BLE_ADVERTISER.start() except Exception as e: print(f"⚠️ [BLE ADVERTISER] Start notice: {e}") server_holder: List[Any] = [None] try: server = await websockets.serve(ws_handler, "0.0.0.0", 8088) server_holder[0] = server print(f"🚀 SimOS iRacing Bridge v{BRIDGE_VERSION} running on ws://0.0.0.0:8088 (Ctrl+C to stop)") # Automatic background synchronization of host PC machine passport to GridPass Cloud threading.Thread(target=sync_to_gridpass, daemon=True).start() except OSError as e: print(f"\n⚠️ [PORT 8088 CONFLICT] Port 8088 is occupied: {e}") print(" Another process or SimOS instance is already running on port 8088.") print(" Entering background recovery mode (will bind automatically when port opens)...") send_tray_notification( "⚠️ Port 8088 Occupied", "Port 8088 is in use by another process. SimOS will auto-bind when freed." ) tasks = [ telemetry_broadcaster(), auto_updater_loop(), BLE_SCANNER.run_loop(), ] if server_holder[0] is not None: tasks.append(server_holder[0].wait_closed() if hasattr(server_holder[0], 'wait_closed') else asyncio.Future()) else: tasks.append(recover_ws_port_loop(server_holder)) results = await asyncio.gather(*tasks, return_exceptions=True) for r in results: if isinstance(r, Exception): print(f"⚠️ [SimOS Task Notice] Background task ended with: {r}") if __name__ == "__main__": # Handle CLI pairing token argument for i, arg in enumerate(sys.argv): if arg in ("--token", "--pairing-token") and i + 1 < len(sys.argv): token_val = sys.argv[i + 1].strip().upper() cfg_dir = os.path.join(os.environ.get("LOCALAPPDATA", ""), "GridPass", "SimOS") os.makedirs(cfg_dir, exist_ok=True) cfg_p = os.path.join(cfg_dir, "config.json") cfg_data = {} if os.path.exists(cfg_p): try: with open(cfg_p, "r", encoding="utf-8") as f: cfg_data = json.load(f) except Exception: pass cfg_data["pairing_token"] = token_val with open(cfg_p, "w", encoding="utf-8") as f: json.dump(cfg_data, f, indent=2) print(f"[SimOS] Saved pairing token to config: {token_val}") # Handle one-shot --sync CLI flag if "--sync" in sys.argv or "--sync-rig" in sys.argv: print(f"🔄 SimOS v{BRIDGE_VERSION} syncing rig to GridPass Cloud...") host_arg = None for i, arg in enumerate(sys.argv): if arg in ("--host", "--url") and i + 1 < len(sys.argv): host_arg = sys.argv[i + 1] res = sync_to_gridpass(host_url=host_arg) if res and res.get("success"): print("✅ SimOS Cloud sync completed successfully.") sys.exit(0) else: print("❌ SimOS Cloud sync could not connect to endpoint.") sys.exit(1) try: asyncio.run(main()) except KeyboardInterrupt: print("\n[SHUTDOWN] SimOS iRacing Bridge stopped.") except BaseException as e: import traceback print(f"\n[FATAL ERROR] {e}") traceback.print_exc() finally: if BLE_ADVERTISER: try: BLE_ADVERTISER.stop() except Exception: pass if TRAY_ICON: try: TRAY_ICON.stop() except Exception: pass