From 3439f546710ecf0c16d7d99eeddd435b8c64750d Mon Sep 17 00:00:00 2001 From: Fasde Date: Mon, 29 Jun 2026 23:02:25 +0200 Subject: [PATCH 1/5] Added Support for LED-to-RPM connection via game telemetry. Initial support for Assetto Corsa Rally via shared memory and Dirt Rally 2 via UDP --- boxflat/panels/presets.py | 2 +- boxflat/panels/wheel_old.py | 144 ++++++++++++++++++- boxflat/telemetry/__init__.py | 0 boxflat/telemetry/assetto_corsa_rally.py | 10 ++ boxflat/telemetry/base_telemetry.py | 172 +++++++++++++++++++++++ boxflat/telemetry/dirt_rally_2.py | 125 ++++++++++++++++ boxflat/telemetry/telemetry_manager.py | 22 +++ 7 files changed, 473 insertions(+), 2 deletions(-) create mode 100644 boxflat/telemetry/__init__.py create mode 100644 boxflat/telemetry/assetto_corsa_rally.py create mode 100644 boxflat/telemetry/base_telemetry.py create mode 100644 boxflat/telemetry/dirt_rally_2.py create mode 100644 boxflat/telemetry/telemetry_manager.py diff --git a/boxflat/panels/presets.py b/boxflat/panels/presets.py index da486e7c..fe42fb96 100644 --- a/boxflat/panels/presets.py +++ b/boxflat/panels/presets.py @@ -195,7 +195,7 @@ def _load_preset(self, preset_name: str, automatic=False, default=False): app = self._application notif.set_title(f"Detected: {pm.get_linked_process()}" if automatic else "No games detected") - notif.set_body(f"Loading {"default" if default else ""} preset: {preset_name}") + notif.set_body(f"Loading {default} if default else ""} preset: {preset_name}") notif.set_priority(NotificationPriority.NORMAL) app.send_notification("preset", notif) diff --git a/boxflat/panels/wheel_old.py b/boxflat/panels/wheel_old.py index f4f93cc1..386f63a1 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,27 @@ 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 +411,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 +542,98 @@ 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: + thresholds = [ + self._cm.get_setting(f"wheel-rpm-value{i+1}", exclusive=True) + for i in range(NUM_LEDS) + ] + else: + 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 + + 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: + 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 + + 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}.") + + 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 + + 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..71dfe41a --- /dev/null +++ b/boxflat/telemetry/base_telemetry.py @@ -0,0 +1,172 @@ +import mmap +import os +import struct + + +class BaseTelemetry: + GAME_NAME = "DEFAULT" + + def __init__(self): + self.GAME_NAME = type(self).GAME_NAME + self.source_name = "" + + def connect(self): + raise NotImplementedError + + def is_connected(self): + return True + + def get_rpm(self): + raise NotImplementedError + + def close(self): + pass + + +class MmapTelemetry(BaseTelemetry): + PHYSICS_PATH = "" + PHYSICS_SIZE = 0 + STATIC_PATH = "" + STATIC_SIZE = 0 + OFFSET_RPM = 0 + OFFSET_CURRENT_MAX_RPM = 0 + OFFSET_STATIC_MAX_RPM = None + RPM_STRUCT = "=i" + MAX_RPM_STRUCT = "=i" + DEFAULT_MAX_RPM = 8000 + + def __init__(self): + super().__init__() + self.phys = None + self.static = None + self._file = None + self._static_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: + 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: + file.close() + raise + + self._file = file + if not self._connect_static(): + self.close() + return False + 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 + + if self.OFFSET_STATIC_MAX_RPM is not None: + return self.static is not None and os.path.exists(self.STATIC_PATH) + + 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] + + if self.OFFSET_STATIC_MAX_RPM is not None: + self.static.seek(self.OFFSET_STATIC_MAX_RPM) + max_rpm = struct.unpack(self.MAX_RPM_STRUCT, self.static.read(4))[0] + else: + 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 _connect_static(self): + if self.OFFSET_STATIC_MAX_RPM is None: + return True + + if self.static is not None: + return True + + if not os.path.exists(self.STATIC_PATH): + return False + + try: + file = open(self.STATIC_PATH, "rb") + if os.fstat(file.fileno()).st_size < self.STATIC_SIZE: + file.close() + return False + + try: + self.static = mmap.mmap( + file.fileno(), + self.STATIC_SIZE, + access=mmap.ACCESS_READ + ) + except OSError: + file.close() + raise + + self._static_file = file + except OSError as e: + print(f"{self.GAME_NAME} static telemetry connect failed: {e}") + return False + + return True + + 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 + + if self.static is not None: + try: + self.static.close() + except (BufferError, OSError): + pass + self.static = None + + if self._static_file is not None: + try: + self._static_file.close() + except OSError: + pass + self._static_file = None diff --git a/boxflat/telemetry/dirt_rally_2.py b/boxflat/telemetry/dirt_rally_2.py new file mode 100644 index 00000000..65c1be8b --- /dev/null +++ b/boxflat/telemetry/dirt_rally_2.py @@ -0,0 +1,125 @@ +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 + + def __init__(self): + super().__init__() + self._udp_socket = None + self._last_rpm = 0 + self._last_max_rpm = 8000 + self._packets_received = 0 + self._packets_rejected = 0 + + # Codemasters/EGO telemetry packet floats: engineRate is rad/s, maxRPM is RPM. + # We have to convert engineRate back to rpm + self.OFFSET_RPM = 37 * 4 + self.OFFSET_CURRENT_MAX_RPM = 61 * 4 + + 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, 8000 + + 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"DiRT Rally 2.0 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"DiRT Rally 2.0 UDP telemetry read failed: {e}") + return 0, 8000 + + if len(packet) < self.PACKET_SIZE: + continue + + self._packets_received += 1 + rpm_data = self._extract_rpm(packet) + + if rpm_data is None: + self._packets_rejected += 1 + 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 8000 + + return self._last_rpm, self._last_max_rpm + + def _engine_rate_to_rpm(self, engine_rate): + 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 + (63 * 4))[0] + max_gears = struct.unpack_from("=f", packet, offset + (65 * 4))[0] + except struct.error: + continue + + 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..5658c5e8 --- /dev/null +++ b/boxflat/telemetry/telemetry_manager.py @@ -0,0 +1,22 @@ +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 From b7c9fb08fceda616a69f8244dff7979ffc0d9c4d Mon Sep 17 00:00:00 2001 From: Fasde Date: Mon, 29 Jun 2026 23:07:25 +0200 Subject: [PATCH 2/5] Redo preset change --- boxflat/panels/presets.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/boxflat/panels/presets.py b/boxflat/panels/presets.py index fe42fb96..da486e7c 100644 --- a/boxflat/panels/presets.py +++ b/boxflat/panels/presets.py @@ -195,7 +195,7 @@ def _load_preset(self, preset_name: str, automatic=False, default=False): app = self._application notif.set_title(f"Detected: {pm.get_linked_process()}" if automatic else "No games detected") - notif.set_body(f"Loading {default} if default else ""} preset: {preset_name}") + notif.set_body(f"Loading {"default" if default else ""} preset: {preset_name}") notif.set_priority(NotificationPriority.NORMAL) app.send_notification("preset", notif) From 93ce5d450445c6af41474f42076fb36a6bc2fd63 Mon Sep 17 00:00:00 2001 From: Fasde <48734014+Fasde@users.noreply.github.com> Date: Tue, 30 Jun 2026 16:18:17 +0200 Subject: [PATCH 3/5] Refactor and Cleanup --- boxflat/telemetry/base_telemetry.py | 10 ++++++++++ boxflat/telemetry/dirt_rally_2.py | 22 +++++++++++++++++----- boxflat/telemetry/telemetry_manager.py | 1 + 3 files changed, 28 insertions(+), 5 deletions(-) diff --git a/boxflat/telemetry/base_telemetry.py b/boxflat/telemetry/base_telemetry.py index 71dfe41a..454ccb5f 100644 --- a/boxflat/telemetry/base_telemetry.py +++ b/boxflat/telemetry/base_telemetry.py @@ -10,15 +10,19 @@ def __init__(self): self.GAME_NAME = type(self).GAME_NAME self.source_name = "" + def connect(self): raise NotImplementedError + def is_connected(self): return True + def get_rpm(self): raise NotImplementedError + def close(self): pass @@ -35,6 +39,7 @@ class MmapTelemetry(BaseTelemetry): MAX_RPM_STRUCT = "=i" DEFAULT_MAX_RPM = 8000 + def __init__(self): super().__init__() self.phys = None @@ -43,6 +48,7 @@ def __init__(self): self._static_file = None self.source_name = self.PHYSICS_PATH + def connect(self): if self.phys is not None: return self.is_connected() @@ -77,6 +83,7 @@ def connect(self): return True + def is_connected(self): if self.phys is None or not os.path.exists(self.PHYSICS_PATH): return False @@ -86,6 +93,7 @@ def is_connected(self): return True + def get_rpm(self): if self.phys is None: return 0, self.DEFAULT_MAX_RPM @@ -109,6 +117,7 @@ def get_rpm(self): return int(rpm), int(max_rpm) + def _connect_static(self): if self.OFFSET_STATIC_MAX_RPM is None: return True @@ -142,6 +151,7 @@ def _connect_static(self): return True + def close(self): if self.phys is not None: try: diff --git a/boxflat/telemetry/dirt_rally_2.py b/boxflat/telemetry/dirt_rally_2.py index 65c1be8b..64565178 100644 --- a/boxflat/telemetry/dirt_rally_2.py +++ b/boxflat/telemetry/dirt_rally_2.py @@ -19,11 +19,15 @@ def __init__(self): self._last_max_rpm = 8000 self._packets_received = 0 self._packets_rejected = 0 + self.FALLBACK_RPM = 8000 # Codemasters/EGO telemetry packet floats: engineRate is rad/s, maxRPM is RPM. # We have to convert engineRate back to rpm self.OFFSET_RPM = 37 * 4 self.OFFSET_CURRENT_MAX_RPM = 61 * 4 + self.IDLE_RPM = 63 * 4 + self.MAX_GEARS = 65 * 4 + def connect(self): if self.is_connected(): @@ -35,14 +39,17 @@ def connect(self): 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, 8000 + return 0, self.FALLBACK_RPM + def close(self): if self._udp_socket is not None: @@ -52,6 +59,7 @@ def close(self): pass self._udp_socket = None + def _ensure_udp_socket(self): if self._udp_socket is not None: return True @@ -69,6 +77,7 @@ def _ensure_udp_socket(self): self._udp_socket = udp_socket return True + def _get_udp_rpm(self): while True: try: @@ -77,7 +86,7 @@ def _get_udp_rpm(self): break except OSError as e: print(f"DiRT Rally 2.0 UDP telemetry read failed: {e}") - return 0, 8000 + return 0, self.FALLBACK_RPM if len(packet) < self.PACKET_SIZE: continue @@ -91,20 +100,23 @@ def _get_udp_rpm(self): 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 8000 + self._last_max_rpm = int(max_rpm) if max_rpm > 0 else self.FALLBACK_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 + (63 * 4))[0] - max_gears = struct.unpack_from("=f", packet, offset + (65 * 4))[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 diff --git a/boxflat/telemetry/telemetry_manager.py b/boxflat/telemetry/telemetry_manager.py index 5658c5e8..18a8f556 100644 --- a/boxflat/telemetry/telemetry_manager.py +++ b/boxflat/telemetry/telemetry_manager.py @@ -15,6 +15,7 @@ def get_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: From 656dcab5b1fc555ed034d189bf2ec4dece70a0af Mon Sep 17 00:00:00 2001 From: Fasde Date: Sat, 4 Jul 2026 08:47:06 +0200 Subject: [PATCH 4/5] Cleanup stuff that was only used in development, some comments --- boxflat/panels/wheel_old.py | 4 ++ boxflat/telemetry/base_telemetry.py | 81 +++++------------------------ boxflat/telemetry/dirt_rally_2.py | 31 +++++++---- 3 files changed, 37 insertions(+), 79 deletions(-) diff --git a/boxflat/panels/wheel_old.py b/boxflat/panels/wheel_old.py index 386f63a1..5f63cde8 100644 --- a/boxflat/panels/wheel_old.py +++ b/boxflat/panels/wheel_old.py @@ -580,6 +580,7 @@ def rpm_to_mask(rpm, max_rpm): 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() @@ -587,6 +588,7 @@ def rpm_to_mask(rpm, max_rpm): 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...") @@ -595,6 +597,7 @@ def rpm_to_mask(rpm, max_rpm): 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 @@ -625,6 +628,7 @@ def rpm_to_mask(rpm, max_rpm): 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: diff --git a/boxflat/telemetry/base_telemetry.py b/boxflat/telemetry/base_telemetry.py index 71dfe41a..ed88de8b 100644 --- a/boxflat/telemetry/base_telemetry.py +++ b/boxflat/telemetry/base_telemetry.py @@ -5,6 +5,7 @@ class BaseTelemetry: GAME_NAME = "DEFAULT" + DEFAULT_MAX_RPM = 8000 def __init__(self): self.GAME_NAME = type(self).GAME_NAME @@ -14,35 +15,31 @@ def connect(self): raise NotImplementedError def is_connected(self): - return True + raise NotImplementedError def get_rpm(self): raise NotImplementedError def close(self): - pass + raise NotImplementedError class MmapTelemetry(BaseTelemetry): PHYSICS_PATH = "" PHYSICS_SIZE = 0 - STATIC_PATH = "" - STATIC_SIZE = 0 OFFSET_RPM = 0 OFFSET_CURRENT_MAX_RPM = 0 - OFFSET_STATIC_MAX_RPM = None RPM_STRUCT = "=i" MAX_RPM_STRUCT = "=i" - DEFAULT_MAX_RPM = 8000 + def __init__(self): super().__init__() self.phys = None - self.static = None self._file = None - self._static_file = None self.source_name = self.PHYSICS_PATH + def connect(self): if self.phys is not None: return self.is_connected() @@ -51,6 +48,7 @@ def connect(self): 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() @@ -62,14 +60,12 @@ def connect(self): self.PHYSICS_SIZE, access=mmap.ACCESS_READ ) - except OSError: + except OSError as e: + print(f"{self.GAME_NAME} telemetry connect failed: {e}") file.close() - raise + raise e self._file = file - if not self._connect_static(): - self.close() - return False except OSError as e: print(f"{self.GAME_NAME} telemetry connect failed: {e}") self.close() @@ -77,15 +73,14 @@ def connect(self): return True + def is_connected(self): if self.phys is None or not os.path.exists(self.PHYSICS_PATH): return False - if self.OFFSET_STATIC_MAX_RPM is not None: - return self.static is not None and os.path.exists(self.STATIC_PATH) - return True + def get_rpm(self): if self.phys is None: return 0, self.DEFAULT_MAX_RPM @@ -94,12 +89,8 @@ def get_rpm(self): self.phys.seek(self.OFFSET_RPM) rpm = struct.unpack(self.RPM_STRUCT, self.phys.read(4))[0] - if self.OFFSET_STATIC_MAX_RPM is not None: - self.static.seek(self.OFFSET_STATIC_MAX_RPM) - max_rpm = struct.unpack(self.MAX_RPM_STRUCT, self.static.read(4))[0] - else: - self.phys.seek(self.OFFSET_CURRENT_MAX_RPM) - max_rpm = struct.unpack(self.MAX_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 @@ -109,38 +100,6 @@ def get_rpm(self): return int(rpm), int(max_rpm) - def _connect_static(self): - if self.OFFSET_STATIC_MAX_RPM is None: - return True - - if self.static is not None: - return True - - if not os.path.exists(self.STATIC_PATH): - return False - - try: - file = open(self.STATIC_PATH, "rb") - if os.fstat(file.fileno()).st_size < self.STATIC_SIZE: - file.close() - return False - - try: - self.static = mmap.mmap( - file.fileno(), - self.STATIC_SIZE, - access=mmap.ACCESS_READ - ) - except OSError: - file.close() - raise - - self._static_file = file - except OSError as e: - print(f"{self.GAME_NAME} static telemetry connect failed: {e}") - return False - - return True def close(self): if self.phys is not None: @@ -156,17 +115,3 @@ def close(self): except OSError: pass self._file = None - - if self.static is not None: - try: - self.static.close() - except (BufferError, OSError): - pass - self.static = None - - if self._static_file is not None: - try: - self._static_file.close() - except OSError: - pass - self._static_file = None diff --git a/boxflat/telemetry/dirt_rally_2.py b/boxflat/telemetry/dirt_rally_2.py index 65c1be8b..35c7e71f 100644 --- a/boxflat/telemetry/dirt_rally_2.py +++ b/boxflat/telemetry/dirt_rally_2.py @@ -10,21 +10,21 @@ class DirtRally2(BaseTelemetry): PACKET_FLOATS = 66 PACKET_SIZE = PACKET_FLOATS * 4 - UDP_PORT = 20777 + UDP_PORT = 20777 # can be set in the DR2.0 config file + def __init__(self): super().__init__() self._udp_socket = None self._last_rpm = 0 - self._last_max_rpm = 8000 - self._packets_received = 0 - self._packets_rejected = 0 + self._last_max_rpm = self.DEFAULT_MAX_RPM # Codemasters/EGO telemetry packet floats: engineRate is rad/s, maxRPM is RPM. # We have to convert engineRate back to rpm self.OFFSET_RPM = 37 * 4 self.OFFSET_CURRENT_MAX_RPM = 61 * 4 + def connect(self): if self.is_connected(): return True @@ -35,14 +35,17 @@ def connect(self): 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, 8000 + return 0, self.DEFAULT_MAX_RPM + def close(self): if self._udp_socket is not None: @@ -52,6 +55,7 @@ def close(self): pass self._udp_socket = None + def _ensure_udp_socket(self): if self._udp_socket is not None: return True @@ -63,12 +67,13 @@ def _ensure_udp_socket(self): udp_socket.setblocking(False) udp_socket.bind(("", port)) except (OSError, ValueError) as e: - print(f"DiRT Rally 2.0 UDP telemetry bind failed: {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: @@ -76,28 +81,28 @@ def _get_udp_rpm(self): except BlockingIOError: break except OSError as e: - print(f"DiRT Rally 2.0 UDP telemetry read failed: {e}") - return 0, 8000 + print(f"{self.GAME_NAME} UDP telemetry read failed: {e}") + return 0, self.DEFAULT_MAX_RPM if len(packet) < self.PACKET_SIZE: continue - self._packets_received += 1 rpm_data = self._extract_rpm(packet) if rpm_data is None: - self._packets_rejected += 1 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 8000 + 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): 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: @@ -108,6 +113,10 @@ def _extract_rpm(self, packet): 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 From ef5fa399480540cb785ba8ce9463de7f6d660da7 Mon Sep 17 00:00:00 2001 From: Fasde Date: Sat, 4 Jul 2026 08:56:45 +0200 Subject: [PATCH 5/5] Bit more commenting --- boxflat/panels/wheel_old.py | 4 ++++ boxflat/telemetry/dirt_rally_2.py | 15 +++++++-------- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/boxflat/panels/wheel_old.py b/boxflat/panels/wheel_old.py index 5f63cde8..84aa8d24 100644 --- a/boxflat/panels/wheel_old.py +++ b/boxflat/panels/wheel_old.py @@ -304,6 +304,7 @@ def prepare_ui(self): 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() @@ -553,11 +554,13 @@ def rpm_to_mask(rpm, max_rpm): 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] @@ -572,6 +575,7 @@ def rpm_to_mask(rpm, max_rpm): 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) diff --git a/boxflat/telemetry/dirt_rally_2.py b/boxflat/telemetry/dirt_rally_2.py index a806969e..3742fe9b 100644 --- a/boxflat/telemetry/dirt_rally_2.py +++ b/boxflat/telemetry/dirt_rally_2.py @@ -12,6 +12,13 @@ class DirtRally2(BaseTelemetry): 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__() @@ -19,14 +26,6 @@ def __init__(self): self._last_rpm = 0 self._last_max_rpm = self.DEFAULT_MAX_RPM - # Codemasters/EGO telemetry packet floats: engineRate is rad/s, maxRPM is RPM. - # We have to convert engineRate back to rpm - self.OFFSET_RPM = 37 * 4 - self.OFFSET_CURRENT_MAX_RPM = 61 * 4 - self.IDLE_RPM = 63 * 4 - self.MAX_GEARS = 65 * 4 - - def connect(self): if self.is_connected():