from __future__ import annotations import configparser from pathlib import Path from typing import Any from .paths import EnvironmentPaths BRIDGE_NAME = "CODEXBRIDGE" BRIDGE_DIR_NAME = "CodexBridge" BRIDGE_APP_INI = """[ABOUT] NAME = Codex Bridge AUTHOR = Codex VERSION = 1.0 DESCRIPTION = Publishes richer live telemetry for external coaching. """ BRIDGE_APP_PY = r'''import ac import acsys import mmap import ctypes import threading class vec3(ctypes.Structure): _pack_ = 4 _fields_ = [('x', ctypes.c_float), ('y', ctypes.c_float), ('z', ctypes.c_float)] class vec4(ctypes.Structure): _pack_ = 4 _fields_ = [('a', ctypes.c_float), ('b', ctypes.c_float), ('c', ctypes.c_float), ('d', ctypes.c_float)] class CodexPage(ctypes.Structure): _pack_ = 4 _fields_ = [ ('packetId', ctypes.c_int32), ('speedMS', ctypes.c_float), ('speedKMH', ctypes.c_float), ('gas', ctypes.c_float), ('brake', ctypes.c_float), ('clutch', ctypes.c_float), ('steer', ctypes.c_float), ('gear', ctypes.c_int32), ('rpm', ctypes.c_int32), ('lapCount', ctypes.c_int32), ('lapInvalidated', ctypes.c_int32), ('lapTime', ctypes.c_int32), ('lastLap', ctypes.c_int32), ('bestLap', ctypes.c_int32), ('performanceMeter', ctypes.c_float), ('lastFF', ctypes.c_float), ('normalizedSplinePosition', ctypes.c_float), ('worldPosition', vec3), ('velocity', vec3), ('localVelocity', vec3), ('localAngularVelocity', vec3), ('accG', vec3), ('load', vec4), ('mz', vec4), ('ndSlip', vec4), ('slipAngle', vec4), ('slipRatio', vec4), ('tyreSlip', vec4), ('coreTemp', vec4), ('camberDeg', vec4), ('suspensionTravel', vec4), ('wheelAngularSpeed', vec4), ('driverName', ctypes.c_char * 64), ('carModel', ctypes.c_char * 64), ] packet = 0 timer = 0 event = threading.Event() stop_thread = False appWindow = 0 sharedPage = None mm = None def _vec3_from_tuple(values): result = vec3() result.x = float(values[0]) result.y = float(values[1]) result.z = float(values[2]) return result def _vec4_from_tuple(values): result = vec4() result.a = float(values[0]) result.b = float(values[1]) result.c = float(values[2]) result.d = float(values[3]) return result def _write_loop(): global sharedPage, stop_thread, packet while not stop_thread: event.wait() event.clear() packet += 1 sharedPage.packetId = packet sharedPage.speedMS = ac.getCarState(0, acsys.CS.SpeedMS) sharedPage.speedKMH = ac.getCarState(0, acsys.CS.SpeedKMH) sharedPage.gas = ac.getCarState(0, acsys.CS.Gas) sharedPage.brake = ac.getCarState(0, acsys.CS.Brake) sharedPage.clutch = ac.getCarState(0, acsys.CS.Clutch) sharedPage.steer = ac.getCarState(0, acsys.CS.Steer) sharedPage.gear = int(ac.getCarState(0, acsys.CS.Gear)) sharedPage.rpm = int(ac.getCarState(0, acsys.CS.RPM)) sharedPage.lapCount = int(ac.getCarState(0, acsys.CS.LapCount)) sharedPage.lapInvalidated = int(ac.getCarState(0, acsys.CS.LapInvalidated)) sharedPage.lapTime = int(ac.getCarState(0, acsys.CS.LapTime)) sharedPage.lastLap = int(ac.getCarState(0, acsys.CS.LastLap)) sharedPage.bestLap = int(ac.getCarState(0, acsys.CS.BestLap)) sharedPage.performanceMeter = ac.getCarState(0, acsys.CS.PerformanceMeter) sharedPage.lastFF = ac.getCarState(0, acsys.CS.LastFF) sharedPage.normalizedSplinePosition = ac.getCarState(0, acsys.CS.NormalizedSplinePosition) sharedPage.worldPosition = _vec3_from_tuple(ac.getCarState(0, acsys.CS.WorldPosition)) sharedPage.velocity = _vec3_from_tuple(ac.getCarState(0, acsys.CS.Velocity)) sharedPage.localVelocity = _vec3_from_tuple(ac.getCarState(0, acsys.CS.LocalVelocity)) sharedPage.localAngularVelocity = _vec3_from_tuple(ac.getCarState(0, acsys.CS.LocalAngularVelocity)) sharedPage.accG = _vec3_from_tuple(ac.getCarState(0, acsys.CS.AccG)) sharedPage.load = _vec4_from_tuple(ac.getCarState(0, acsys.CS.Load)) sharedPage.mz = _vec4_from_tuple(ac.getCarState(0, acsys.CS.Mz)) sharedPage.ndSlip = _vec4_from_tuple(ac.getCarState(0, acsys.CS.NdSlip)) sharedPage.slipAngle = _vec4_from_tuple(ac.getCarState(0, acsys.CS.SlipAngle)) sharedPage.slipRatio = _vec4_from_tuple(ac.getCarState(0, acsys.CS.SlipRatio)) sharedPage.tyreSlip = _vec4_from_tuple(ac.getCarState(0, acsys.CS.TyreSlip)) sharedPage.coreTemp = _vec4_from_tuple(ac.getCarState(0, acsys.CS.CurrentTyresCoreTemp)) sharedPage.camberDeg = _vec4_from_tuple(ac.getCarState(0, acsys.CS.CamberDeg)) sharedPage.suspensionTravel = _vec4_from_tuple(ac.getCarState(0, acsys.CS.SuspensionTravel)) sharedPage.wheelAngularSpeed = _vec4_from_tuple(ac.getCarState(0, acsys.CS.WheelAngularSpeed)) sharedPage.driverName = ac.getDriverName(0).encode('utf-8') sharedPage.carModel = ac.getCarName(0).encode('utf-8') def acMain(ac_version): global appWindow, sharedPage, mm appWindow = ac.newApp("Codex Bridge") ac.setSize(appWindow, 1, 1) ac.drawBorder(appWindow, 0) ac.setBackgroundOpacity(appWindow, 0) mm = mmap.mmap(0, ctypes.sizeof(CodexPage), "acpmf_Codex") sharedPage = CodexPage.from_buffer(mm) thread = threading.Thread(target=_write_loop) thread.daemon = True thread.start() return "Codex Bridge" def acUpdate(deltaT): global timer timer += deltaT if timer > 0.025: event.set() timer = 0 def acShutdown(): global stop_thread stop_thread = True event.set() ''' def install_codex_bridge_app(paths: EnvironmentPaths, *, enable: bool = True) -> dict[str, Any]: if paths.assetto_root is None: raise RuntimeError("Assetto Corsa install was not found") app_root = paths.assetto_root / "apps" / "python" / BRIDGE_DIR_NAME app_root.mkdir(parents=True, exist_ok=True) ini_path = app_root / "app.ini" py_path = app_root / f"{BRIDGE_DIR_NAME}.py" ini_path.write_text(BRIDGE_APP_INI, encoding="utf-8") py_path.write_text(BRIDGE_APP_PY, encoding="utf-8") python_ini = paths.cfg_root / "python.ini" parser = configparser.ConfigParser() parser.optionxform = str if python_ini.exists(): parser.read(python_ini, encoding="utf-8") if not parser.has_section(BRIDGE_NAME): parser.add_section(BRIDGE_NAME) parser.set(BRIDGE_NAME, "ACTIVE", "1" if enable else "0") with python_ini.open("w", encoding="utf-8") as handle: parser.write(handle, space_around_delimiters=False) return { "app_root": str(app_root), "ini_path": str(ini_path), "py_path": str(py_path), "python_ini": str(python_ini), "enabled": enable, }