diff --git a/boxflat/panels/wheel_old.py b/boxflat/panels/wheel_old.py index f4f93cc1..84aa8d24 100644 --- a/boxflat/panels/wheel_old.py +++ b/boxflat/panels/wheel_old.py @@ -1,10 +1,13 @@ # Copyright (c) 2025, Tomasz PakuĊ‚a Using Arch BTW +import time +from threading import Thread + from boxflat.panels.settings_panel import SettingsPanel from boxflat.connection_manager import MozaConnectionManager from boxflat.bitwise import * +from boxflat.telemetry.telemetry_manager import Telemetries, telemetry_from_game_name from boxflat.widgets import * - from boxflat.hid_handler import MozaAxis from boxflat.settings_handler import SettingsHandler @@ -57,6 +60,7 @@ class OldWheelSettings(SettingsPanel): def __init__(self, button_callback, connection_manager: MozaConnectionManager, hid_handler, settings: SettingsHandler): self._settings = settings self._blinking_row = None + self._telemetry_row = None self._split = None self._timing_row = None @@ -80,6 +84,7 @@ def __init__(self, button_callback, connection_manager: MozaConnectionManager, h super().__init__("Wheel Old", button_callback, connection_manager, hid_handler) self._cm.subscribe_connected("wheel-rpm-value1", self.active) self.set_banner_title(f"Device disconnected...") + self.telemetry = None def active(self, value: int): @@ -287,6 +292,28 @@ def prepare_ui(self): self._add_row(BoxflatButtonRow("Wheel indicator test", "Test")) self._current_row.subscribe(self.start_test) + self.add_preferences_group() + self._telemetry_row = BoxflatComboRow("Choose game", "Choose game") + self._add_row(self._telemetry_row) + self._telemetry_row.get_model().append("") + self._telemetry_row.add_entries(*[t.value.GAME_NAME for t in Telemetries]) + self._telemetry_row.subscribe(self._change_telemetry, self._telemetry_row.get_selected_item) + + + self._add_row(BoxflatButtonRow("RPM to LED Connection", "Start")) + self._current_row.subscribe(self.start_led_rpm_connection) + self._current_row.add_button("Stop", self.stop_led_rpm_connection) + + + def _change_telemetry(self, value, func): + telemetry_string = func().get_string() + + telemetry = telemetry_from_game_name(telemetry_string) + if telemetry is None: + return + + self.telemetry = telemetry + def _set_rpm_timings(self, timings: list): self._cm.set_setting(timings, "wheel-rpm-timings") @@ -385,6 +412,30 @@ def start_test(self, *args): self._test_thread = Thread(daemon=True, target=self._wheel_rpm_test).start() + def start_led_rpm_connection(self, *args): + if self.telemetry is None: + print("No telemetry game selected.") + return + + if getattr(self, "_led_rpm_running", False): + print("RPM bridge already running") + return + + self._led_rpm_running = True + self._led_thread = Thread( + daemon=True, + target=self._wheel_led_rpm_connection + ) + self._led_thread.start() + + def stop_led_rpm_connection(self, *args): + self._led_rpm_running = False + if self.telemetry is not None: + self.telemetry.close() + time.sleep(0.1) + self.reset() + + def _sync_from_dash(self, *args): for setting in SHARED_SETTINGS: value = self._cm.get_setting(f"dash-{setting}", exclusive=True) @@ -492,6 +543,105 @@ def _wheel_rpm_test(self, *args): self._cm.set_setting(initial_mode, "wheel-rpm-indicator-mode", exclusive=True) + def _wheel_led_rpm_connection(self, *args): + NUM_LEDS = 10 + UPDATE_RATE = 1 / 50 + + def rpm_to_mask(rpm, max_rpm): + if rpm <= 0 or max_rpm <= 0: + return 0 + + rpm_mode = self._cm.get_setting("wheel-rpm-mode", exclusive=True) + + if rpm_mode == 1: + # Fixed RPM mode: read the rpm thresholds set in boxflat UI + thresholds = [ + self._cm.get_setting(f"wheel-rpm-value{i+1}", exclusive=True) + for i in range(NUM_LEDS) + ] + else: + # Percentage RPM mode: calculate the rpm thresholds based on current maximum rpm + timings = self._cm.get_setting("wheel-rpm-timings", exclusive=True) + if timings is None or len(timings) < NUM_LEDS: + timings = self._timings[0] + + thresholds = [ + max_rpm * timing / 100 + for timing in timings[:NUM_LEDS] + ] + + leds = 0 + for threshold in thresholds: + if threshold is not None and rpm >= threshold: + leds += 1 + + # Make each element 1 for each LED that needs to activate + return (1 << leds) - 1 if leds > 0 else 0 + + initial_mode = self._cm.get_setting("wheel-rpm-indicator-mode", exclusive=True) + + last_mode_refresh = 0 + last_debug_print = 0 + + try: + # Outer loop for connecting to the telemetry and retrying if it is not there/fails/disconnects + while self._led_rpm_running: + # Keep wheel in RPM indicator mode + now = time.monotonic() + if now - last_mode_refresh > 2.0: + self._cm.set_setting(1, "wheel-rpm-indicator-mode") + last_mode_refresh = now + + # Wait for active game connection + if not self.telemetry.connect(): + self._cm.set_setting(0, "wheel-old-send-telemetry") + print(f"Waiting for {self.telemetry.GAME_NAME} telemetry...") + time.sleep(1) + continue + + print(f"Connected to {self.telemetry.GAME_NAME} telemetry at {self.telemetry.source_name}.") + + # Inner loop for sending data to wheel once telemetry is connected + while self._led_rpm_running: + try: + # Re-check mode every 2 seconds + now = time.monotonic() + if now - last_mode_refresh > 2.0: + self._cm.set_setting(1, "wheel-rpm-indicator-mode") + last_mode_refresh = now + + if not self.telemetry.is_connected(): + print("Telemetry source disappeared; waiting again.") + self.telemetry.close() + break + + rpm, max_rpm = self.telemetry.get_rpm() + + mask = rpm_to_mask(rpm, max_rpm) + self._cm.set_setting(mask, "wheel-old-send-telemetry") + + now = time.monotonic() + if now - last_debug_print > 1.0: + print(f"RPM telemetry: rpm={rpm} max_rpm={max_rpm} mask={mask}") + last_debug_print = now + + time.sleep(UPDATE_RATE) + + except Exception as e: + print(f"RPM telemetry bridge retrying: {e}") + self._cm.set_setting(0, "wheel-old-send-telemetry") + self.telemetry.close() + time.sleep(1) + # Break out of inner loop so that the outer loop retries connecting + break + + finally: + self._cm.set_setting(0, "wheel-old-send-telemetry") + self._cm.set_setting(initial_mode, "wheel-rpm-indicator-mode", exclusive=True) + if self.telemetry is not None: + self.telemetry.close() + + def reset(self, *_) -> None: self._set_rpm_timings_preset(0) self._set_rpm_timings2_preset(0) diff --git a/boxflat/telemetry/__init__.py b/boxflat/telemetry/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/boxflat/telemetry/assetto_corsa_rally.py b/boxflat/telemetry/assetto_corsa_rally.py new file mode 100644 index 00000000..a6829f1f --- /dev/null +++ b/boxflat/telemetry/assetto_corsa_rally.py @@ -0,0 +1,10 @@ +from boxflat.telemetry.base_telemetry import MmapTelemetry + + +class AssettoCorsaRally(MmapTelemetry): + GAME_NAME = "Assetto Corsa Rally" + PHYSICS_PATH = "/dev/shm/acpmf_physics" # mirror by e.g. DataLink needed + PHYSICS_SIZE = 800 + + OFFSET_RPM = 20 + OFFSET_CURRENT_MAX_RPM = 588 diff --git a/boxflat/telemetry/base_telemetry.py b/boxflat/telemetry/base_telemetry.py new file mode 100644 index 00000000..bdeeb3af --- /dev/null +++ b/boxflat/telemetry/base_telemetry.py @@ -0,0 +1,121 @@ +import mmap +import os +import struct + + +class BaseTelemetry: + GAME_NAME = "DEFAULT" + DEFAULT_MAX_RPM = 8000 + + def __init__(self): + self.GAME_NAME = type(self).GAME_NAME + self.source_name = "" + + + def connect(self): + raise NotImplementedError + + + def is_connected(self): + raise NotImplementedError + + + def get_rpm(self): + raise NotImplementedError + + + def close(self): + raise NotImplementedError + + +class MmapTelemetry(BaseTelemetry): + PHYSICS_PATH = "" + PHYSICS_SIZE = 0 + OFFSET_RPM = 0 + OFFSET_CURRENT_MAX_RPM = 0 + RPM_STRUCT = "=i" + MAX_RPM_STRUCT = "=i" + + + def __init__(self): + super().__init__() + self.phys = None + self._file = None + self.source_name = self.PHYSICS_PATH + + + def connect(self): + if self.phys is not None: + return self.is_connected() + + if not os.path.exists(self.PHYSICS_PATH): + return False + + try: + # No context manager because we close the file in the close function, so that we can continue reading + file = open(self.PHYSICS_PATH, "rb") + if os.fstat(file.fileno()).st_size < self.PHYSICS_SIZE: + file.close() + return False + + try: + self.phys = mmap.mmap( + file.fileno(), + self.PHYSICS_SIZE, + access=mmap.ACCESS_READ + ) + except OSError as e: + print(f"{self.GAME_NAME} telemetry connect failed: {e}") + file.close() + raise e + + self._file = file + except OSError as e: + print(f"{self.GAME_NAME} telemetry connect failed: {e}") + self.close() + return False + + return True + + + def is_connected(self): + if self.phys is None or not os.path.exists(self.PHYSICS_PATH): + return False + + return True + + + def get_rpm(self): + if self.phys is None: + return 0, self.DEFAULT_MAX_RPM + + try: + self.phys.seek(self.OFFSET_RPM) + rpm = struct.unpack(self.RPM_STRUCT, self.phys.read(4))[0] + + self.phys.seek(self.OFFSET_CURRENT_MAX_RPM) + max_rpm = struct.unpack(self.MAX_RPM_STRUCT, self.phys.read(4))[0] + except (ValueError, BufferError, OSError) as e: + print(f"{self.GAME_NAME} telemetry read failed: {e}") + return 0, self.DEFAULT_MAX_RPM + + if max_rpm <= 0: + max_rpm = self.DEFAULT_MAX_RPM + + return int(rpm), int(max_rpm) + + + def close(self): + if self.phys is not None: + try: + self.phys.close() + except (BufferError, OSError): + pass + self.phys = None + + if self._file is not None: + try: + self._file.close() + except OSError: + pass + self._file = None diff --git a/boxflat/telemetry/dirt_rally_2.py b/boxflat/telemetry/dirt_rally_2.py new file mode 100644 index 00000000..3742fe9b --- /dev/null +++ b/boxflat/telemetry/dirt_rally_2.py @@ -0,0 +1,137 @@ +import math +import socket +import struct + +from boxflat.telemetry.base_telemetry import BaseTelemetry + + +class DirtRally2(BaseTelemetry): + GAME_NAME = "DiRT Rally 2.0" + + PACKET_FLOATS = 66 + PACKET_SIZE = PACKET_FLOATS * 4 + UDP_PORT = 20777 # can be set in the DR2.0 config file + + # Codemasters/EGO telemetry packet floats: engineRate is rad/s, maxRPM is RPM. + # We have to convert engineRate back to rpm + OFFSET_RPM = 37 * 4 + OFFSET_CURRENT_MAX_RPM = 61 * 4 + IDLE_RPM = 63 * 4 + MAX_GEARS = 65 * 4 + + + def __init__(self): + super().__init__() + self._udp_socket = None + self._last_rpm = 0 + self._last_max_rpm = self.DEFAULT_MAX_RPM + + + def connect(self): + if self.is_connected(): + return True + + if self._ensure_udp_socket(): + self.source_name = f"udp://0.0.0.0:{self.UDP_PORT}" + return True + + return False + + + def is_connected(self): + return self._udp_socket is not None + + + def get_rpm(self): + if self._udp_socket is not None: + return self._get_udp_rpm() + + return 0, self.DEFAULT_MAX_RPM + + + def close(self): + if self._udp_socket is not None: + try: + self._udp_socket.close() + except OSError: + pass + self._udp_socket = None + + + def _ensure_udp_socket(self): + if self._udp_socket is not None: + return True + + try: + port = self.UDP_PORT + udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + udp_socket.setblocking(False) + udp_socket.bind(("", port)) + except (OSError, ValueError) as e: + print(f"{self.GAME_NAME} UDP telemetry bind failed: {e}") + return False + + self._udp_socket = udp_socket + return True + + + def _get_udp_rpm(self): + while True: + try: + packet = self._udp_socket.recv(4096) + except BlockingIOError: + break + except OSError as e: + print(f"{self.GAME_NAME} UDP telemetry read failed: {e}") + return 0, self.DEFAULT_MAX_RPM + + if len(packet) < self.PACKET_SIZE: + continue + + rpm_data = self._extract_rpm(packet) + + if rpm_data is None: + continue + + rpm, max_rpm = rpm_data + self._last_rpm = int(self._engine_rate_to_rpm(rpm)) + self._last_max_rpm = int(max_rpm) if max_rpm > 0 else self.DEFAULT_MAX_RPM + + return self._last_rpm, self._last_max_rpm + + + def _engine_rate_to_rpm(self, engine_rate): + # convert Radians back to RPM + return engine_rate * 60 / (2 * math.pi) + + + def _extract_rpm(self, packet): + for offset in range(0, len(packet) - self.PACKET_SIZE + 1, 4): + try: + rpm = struct.unpack_from("=f", packet, offset + self.OFFSET_RPM)[0] + max_rpm = struct.unpack_from("=f", packet, offset + self.OFFSET_CURRENT_MAX_RPM)[0] + idle_rpm = struct.unpack_from("=f", packet, offset + self.IDLE_RPM)[0] + max_gears = struct.unpack_from("=f", packet, offset + self.MAX_GEARS)[0] + except struct.error: + continue + + # If any of the following 4 checks fail, we are not reading the data at the correct spot in the file, so we + # try at the next spot + # So we can use these 4 checks (with values we dont need further on) to validate that we are reading the + # rpm data at the correct spot and not read anything that looks rpm-y but isnt + if not 0 <= rpm <= 20000: + continue + + if not 1000 <= max_rpm <= 25000: + continue + + if not 0 <= idle_rpm <= 5000: + continue + + if not 1 <= max_gears <= 12: + continue + + return rpm, max_rpm + + return None diff --git a/boxflat/telemetry/telemetry_manager.py b/boxflat/telemetry/telemetry_manager.py new file mode 100644 index 00000000..18a8f556 --- /dev/null +++ b/boxflat/telemetry/telemetry_manager.py @@ -0,0 +1,23 @@ +import enum + +from boxflat.telemetry.assetto_corsa_rally import AssettoCorsaRally +from boxflat.telemetry.dirt_rally_2 import DirtRally2 + + +class Telemetries(enum.Enum): + acr = AssettoCorsaRally + dr2 = DirtRally2 + + +def get_telemetries(): + telemetries = [] + for telemetry in Telemetries: + telemetries.append((telemetry.name, telemetry.value)) + return telemetries + + +def telemetry_from_game_name(game_name: str): + for telemetry in Telemetries: + if telemetry.value.GAME_NAME == game_name: + return telemetry.value() + return None