From 402a6c4df246bc035dfad8e77022d5dde5e40c83 Mon Sep 17 00:00:00 2001 From: Shane Smiskol Date: Wed, 24 Sep 2025 20:45:56 -0700 Subject: [PATCH] fix toggle callback - also not for naught --- system/ui/lib/wifi_manager.py | 44 ++++++++++++++++++++++++++++++---- system/ui/widgets/list_view.py | 10 +++++++- system/ui/widgets/network.py | 29 +++++++++++++++------- 3 files changed, 69 insertions(+), 14 deletions(-) diff --git a/system/ui/lib/wifi_manager.py b/system/ui/lib/wifi_manager.py index dff3c61678..0716deeba6 100644 --- a/system/ui/lib/wifi_manager.py +++ b/system/ui/lib/wifi_manager.py @@ -15,6 +15,7 @@ from jeepney.low_level import MessageType from jeepney.wrappers import Properties from openpilot.common.swaglog import cloudlog +from openpilot.common.params import Params from openpilot.system.ui.lib.networkmanager import (NM, NM_WIRELESS_IFACE, NM_802_11_AP_SEC_PAIR_WEP40, NM_802_11_AP_SEC_PAIR_WEP104, NM_802_11_AP_SEC_GROUP_WEP40, NM_802_11_AP_SEC_GROUP_WEP104, NM_802_11_AP_SEC_KEY_MGMT_PSK, @@ -136,6 +137,11 @@ class WifiManager: self._last_network_update: float = 0.0 self._callback_queue: list[Callable] = [] + self._tethering_ssid = "weedle" + dongle_id = Params().get("DongleId") + if dongle_id: + self._tethering_ssid += "-" + dongle_id[:4] + # Callbacks self._need_auth: list[Callable[[str], None]] = [] self._activated: list[Callable[[], None]] = [] @@ -153,11 +159,11 @@ class WifiManager: atexit.register(self.stop) - def set_callbacks(self, need_auth: Callable[[str], None] | None, - activated: Callable[[], None] | None, - forgotten: Callable[[], None] | None, - networks_updated: Callable[[list[Network]], None] | None, - disconnected: Callable[[], None] | None): + def set_callbacks(self, need_auth: Callable[[str], None] | None = None, + activated: Callable[[], None] | None = None, + forgotten: Callable[[], None] | None = None, + networks_updated: Callable[[list[Network]], None] | None = None, + disconnected: Callable[[], None] | None = None): if need_auth is not None: self._need_auth.append(need_auth) if activated is not None: @@ -373,6 +379,34 @@ class WifiManager: else: threading.Thread(target=worker, daemon=True).start() + def _deactivate_connection(self, ssid: str): + for conn_path in self._get_active_connections(): + conn_addr = DBusAddress(conn_path, bus_name=NM, interface=NM_ACTIVE_CONNECTION_IFACE) + specific_obj_path = self._router_main.send_and_get_reply(Properties(conn_addr).get('SpecificObject')).body[0][1] + + if specific_obj_path != "/": + ap_addr = DBusAddress(specific_obj_path, bus_name=NM, interface=NM_ACCESS_POINT_IFACE) + ap_ssid = bytes(self._router_main.send_and_get_reply(Properties(ap_addr).get('Ssid')).body[0][1]).decode("utf-8", "replace") + + if ap_ssid == ssid: + self._router_main.send_and_get_reply(new_method_call(self._nm, 'DeactivateConnection', 'o', (conn_path,))) + return + + def is_tethering_active(self) -> bool: + for network in self._networks: + if network.is_connected: + return bool(network.ssid == self._tethering_ssid) + return False + + def set_tethering_active(self, active: bool): + def worker(): + if active: + self.activate_connection(self._tethering_ssid, block=True) + else: + self._deactivate_connection(self._tethering_ssid) + + threading.Thread(target=worker, daemon=True).start() + def _request_scan(self): if self._wifi_device is None: cloudlog.warning("No WiFi device found") diff --git a/system/ui/widgets/list_view.py b/system/ui/widgets/list_view.py index 5706140af8..8e7b182774 100644 --- a/system/ui/widgets/list_view.py +++ b/system/ui/widgets/list_view.py @@ -41,6 +41,9 @@ class ItemAction(Widget, ABC): self.set_rect(rl.Rectangle(0, 0, width, 0)) self._enabled_source = enabled + def set_enabled(self, enabled: bool | Callable[[], bool]): + self._enabled_source = enabled + @property def enabled(self): return _resolve_value(self._enabled_source, False) @@ -51,6 +54,7 @@ class ToggleAction(ItemAction): super().__init__(width, enabled) self.toggle = Toggle(initial_state=initial_state) self.state = initial_state + self._prev_state = initial_state def set_touch_valid_callback(self, touch_callback: Callable[[], bool]) -> None: super().set_touch_valid_callback(touch_callback) @@ -59,7 +63,11 @@ class ToggleAction(ItemAction): def _render(self, rect: rl.Rectangle) -> bool: self.toggle.set_enabled(self.enabled) self.toggle.render(rl.Rectangle(rect.x, rect.y + (rect.height - TOGGLE_HEIGHT) / 2, self._rect.width, TOGGLE_HEIGHT)) - return False + + self.state = self.toggle.get_state() + changed = self.state != self._prev_state + self._prev_state = self.state + return changed def set_state(self, state: bool): self.state = state diff --git a/system/ui/widgets/network.py b/system/ui/widgets/network.py index b4af5c54d7..61264e56d7 100644 --- a/system/ui/widgets/network.py +++ b/system/ui/widgets/network.py @@ -12,7 +12,7 @@ from openpilot.system.ui.widgets.confirm_dialog import ConfirmDialog from openpilot.system.ui.widgets.keyboard import Keyboard from openpilot.system.ui.widgets.label import TextAlignment, gui_label from openpilot.system.ui.widgets.scroller import Scroller -from openpilot.system.ui.widgets.list_view import text_item, button_item, dual_button_item, TextAction, ListItem +from openpilot.system.ui.widgets.list_view import toggle_item, text_item, button_item, dual_button_item, TextAction, ListItem, ToggleAction NM_DEVICE_STATE_NEED_AUTH = 60 MIN_PASSWORD_LENGTH = 8 @@ -113,21 +113,34 @@ class AdvancedNetworkSettings(Widget): # action = - self._ip_address = text_item("IP Address", lambda: self._wifi_manager.ipv4_address) + # enable tethering, tethering password, ~ip address~, wifi network metered, hidden network - # enable tethering, tethering password, ip address, wifi network metered, hidden network + # # tethering = toggle_item("Enable Tethering", initial_state=wifi_manager + # tethering = toggle_item("Enable Tethering", callback=self._wifi_manager)#, initial_state=wifi_manager + + self._tethering_action = ToggleAction(initial_state=False, enabled=True) + self._tethering_btn = ListItem(title="Enable Tethering", action_item=self._tethering_action, callback=self._tethering_toggled) items = [ - self._ip_address + self._tethering_btn, + text_item("IP Address", lambda: self._wifi_manager.ipv4_address) ] self._scroller = Scroller(items, line_separator=True, spacing=0) - # self._wifi_manager.set_callbacks(networks_updated=self._on_network_updated) + self._wifi_manager.set_callbacks(networks_updated=self._on_network_updated) - # def _on_network_updated(self, networks: list[Network]): - # ip_address = self._wifi_manager.ipv4_address - # self._ip_address.title = ip_address if ip_address else "N/A" + def _on_network_updated(self, networks: list[Network]): + self._tethering_action.set_enabled(True) + # ip_address = self._wifi_manager.ipv4_address + # self._ip_address.title = ip_address if ip_address else "N/A" + + def _tethering_toggled(self): + checked = self._tethering_action.state + print("Tethering toggled:", checked) + if checked: + self._tethering_action.set_enabled(False) + self._wifi_manager.set_tethering_active(checked) def _initialize_items(self): items = [