from __future__ import annotations import configparser import json import ctypes import xml.etree.ElementTree as ET from ctypes import wintypes from pathlib import Path from typing import Any from .paths import EnvironmentPaths def _read_json(path: Path) -> dict[str, Any] | None: if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def _read_ini(path: Path) -> configparser.ConfigParser | None: if not path.exists(): return None parser = configparser.ConfigParser() parser.read(path, encoding="utf-8") return parser def _xml_game_entry(path: Path, game_name: str) -> dict[str, Any] | None: if not path.exists(): return None tree = ET.parse(path) root = tree.getroot() for game in root.findall("Game"): name = (game.findtext("name") or "").strip() if name != game_name: continue return { "name": name, "rank": int(game.findtext("rank") or 0), "path": (game.findtext("path") or "").strip() or None, "program": (game.findtext("program") or "").strip() or None, } return None def _setup_kernel32() -> Any: kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.OpenFileMappingW.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.LPCWSTR] kernel32.OpenFileMappingW.restype = wintypes.HANDLE kernel32.MapViewOfFile.argtypes = [ wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD, ctypes.c_size_t, ] kernel32.MapViewOfFile.restype = wintypes.LPVOID kernel32.UnmapViewOfFile.argtypes = [wintypes.LPCVOID] kernel32.UnmapViewOfFile.restype = wintypes.BOOL kernel32.CloseHandle.argtypes = [wintypes.HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL return kernel32 kernel32 = _setup_kernel32() FILE_MAP_READ = 0x0004 class Vec3(ctypes.Structure): _pack_ = 4 _fields_ = [("x", ctypes.c_float), ("y", ctypes.c_float), ("z", ctypes.c_float)] class MozaVehicleInfo(ctypes.Structure): _pack_ = 4 _fields_ = [ ("carId", ctypes.c_int32), ("driverName", ctypes.c_char * 64), ("carModel", ctypes.c_char * 64), ("speedMS", ctypes.c_float), ("bestLapMS", ctypes.c_int32), ("lapCount", ctypes.c_int32), ("currentLapInvalid", ctypes.c_int32), ("currentLapTimeMS", ctypes.c_int32), ("lastLapTimeMS", ctypes.c_int32), ("worldPosition", Vec3), ("isCarInPitline", ctypes.c_int32), ("isCarInPit", ctypes.c_int32), ("carLeaderboardPosition", ctypes.c_int32), ("carRealTimeLeaderboardPosition", ctypes.c_int32), ("spLineLength", ctypes.c_float), ("isConnected", ctypes.c_int32), ] class MozaPage(ctypes.Structure): _pack_ = 4 _fields_ = [ ("numVehicles", ctypes.c_int32), ("focusVehicle", ctypes.c_int32), ("serverName", ctypes.c_char * 512), ("vehicleInfo", MozaVehicleInfo * 64), ("acInstallPath", ctypes.c_char * 512), ("isInternalMemoryModuleLoaded", ctypes.c_int32), ("pluginVersion", ctypes.c_char * 32), ] def _decode_cstr(buffer: Any) -> str: return bytes(buffer).split(b"\x00", 1)[0].decode("utf-8", errors="replace") def read_moza_shared_memory() -> dict[str, Any] | None: handle = kernel32.OpenFileMappingW(FILE_MAP_READ, False, "acpmf_Moza") if not handle: return None view = kernel32.MapViewOfFile(handle, FILE_MAP_READ, 0, 0, 0) if not view: kernel32.CloseHandle(handle) return None try: raw = ctypes.string_at(view, ctypes.sizeof(MozaPage)) page = MozaPage.from_buffer_copy(raw) focus = int(page.focusVehicle) focus_vehicle = None if 0 <= focus < 64: candidate = page.vehicleInfo[focus] focus_vehicle = { "car_id": int(candidate.carId), "driver_name": _decode_cstr(candidate.driverName), "car_model": _decode_cstr(candidate.carModel), "speed_mps": float(candidate.speedMS), "best_lap_ms": int(candidate.bestLapMS), "lap_count": int(candidate.lapCount), "current_lap_invalid": bool(candidate.currentLapInvalid), "current_lap_time_ms": int(candidate.currentLapTimeMS), "last_lap_time_ms": int(candidate.lastLapTimeMS), "world_position": { "x": float(candidate.worldPosition.x), "y": float(candidate.worldPosition.y), "z": float(candidate.worldPosition.z), }, "in_pitline": bool(candidate.isCarInPitline), "in_pit": bool(candidate.isCarInPit), "leaderboard_position": int(candidate.carLeaderboardPosition), "realtime_position": int(candidate.carRealTimeLeaderboardPosition), "spline_position": float(candidate.spLineLength), "is_connected": bool(candidate.isConnected), } return { "num_vehicles": int(page.numVehicles), "focus_vehicle_index": focus, "server_name": _decode_cstr(page.serverName), "ac_install_path": _decode_cstr(page.acInstallPath), "is_internal_memory_module_loaded": bool(page.isInternalMemoryModuleLoaded), "plugin_version": _decode_cstr(page.pluginVersion), "focus_vehicle": focus_vehicle, } finally: kernel32.UnmapViewOfFile(ctypes.c_void_p(view)) kernel32.CloseHandle(handle) def moza_state(paths: EnvironmentPaths) -> dict[str, Any]: moza_docs = paths.documents_root.parent / "MOZA Pit House" if not moza_docs.exists(): return {"available": False} game_config = _xml_game_entry(moza_docs / "GameConfigInfo.xml", "Assetto Corsa") preset_config = _read_ini(moza_docs / "Presets" / "config.ini") last_used = {} if preset_config is not None and preset_config.has_section("LastUsedPreset"): last_used = dict(preset_config.items("LastUsedPreset")) wheelbase_id = last_used.get("6d706d5cea8c8d23") steering_id = last_used.get("79ee031acfc5b674") pedals_id = last_used.get("c0b9e33d07f83788") wheelbase_preset = _read_json(moza_docs / "Presets" / "Motor" / f"{wheelbase_id}.json") if wheelbase_id else None steering_preset = _read_json(moza_docs / "Presets" / "Steering Wheel" / f"{steering_id}.json") if steering_id else None pedals_preset = _read_json(moza_docs / "Presets" / "Pedals" / f"{pedals_id}.json") if pedals_id else None wheelbase_local = _read_json(moza_docs / "LocalParameters" / "WheelBase" / "2c0031001651353032353130.json") pedals_local = _read_json(moza_docs / "LocalParameters" / "Pedals" / "2c00310016513530323531302d5042.json") steering_local = _read_json(moza_docs / "LocalParameters" / "Steering" / "ES.json") state = { "available": True, "documents_root": str(moza_docs), "assetto_corsa_game_config": game_config, "last_used_preset_ids": { "wheelbase": wheelbase_id, "steering_wheel": steering_id, "pedals": pedals_id, }, "wheelbase_preset": wheelbase_preset, "steering_preset": steering_preset, "pedals_preset": pedals_preset, "wheelbase_local": wheelbase_local, "steering_local": steering_local, "pedals_local": pedals_local, "shared_memory": read_moza_shared_memory(), } recommendations: list[str] = [] if game_config and game_config.get("program") == "AssettoCorsa.exe": recommendations.append( "MOZA’s registered AC executable is AssettoCorsa.exe. On this machine the stock launcher has crashed before, so CM or acs.exe is safer for automation." ) if wheelbase_preset: params = wheelbase_preset.get("deviceParams", {}) angle = params.get("maximumSteeringAngle") game_angle = params.get("maximumGameSteeringAngle") ffb = params.get("gameForceFeedbackStrength") if angle is not None and game_angle is not None and angle != game_angle: recommendations.append( f"Wheelbase preset uses different wheel/game angles ({angle}/{game_angle}). Check lock matching if steering feels odd." ) if ffb is not None: recommendations.append(f"MOZA wheelbase AC FFB strength preset is {ffb}.") state["recommendations"] = recommendations return state