diff --git a/files/usr/share/cinnamon/cinnamon-settings/modules/cs_thunderbolt.py b/files/usr/share/cinnamon/cinnamon-settings/modules/cs_thunderbolt.py index 12ce8602ff..4377c57b30 100644 --- a/files/usr/share/cinnamon/cinnamon-settings/modules/cs_thunderbolt.py +++ b/files/usr/share/cinnamon/cinnamon-settings/modules/cs_thunderbolt.py @@ -1,7 +1,5 @@ #!/usr/bin/python3 -import os - import gi gi.require_version("Gtk", "3.0") gi.require_version("Gio", "2.0") @@ -14,6 +12,8 @@ BOLT_BUS_NAME = "org.freedesktop.bolt" BOLT_OBJECT_PATH = "/org/freedesktop/bolt" +BOLT_MANAGER_IFACE = "org.freedesktop.bolt1.Manager" +BOLT_DEVICE_IFACE = "org.freedesktop.bolt1.Device" def build_detail_row(key, value): @@ -26,7 +26,7 @@ def build_detail_row(key, value): else: labelValue = Gtk.Label(label=str(value)) labelValue.set_selectable(True) - labelValue.set_line_wrap(True) + labelValue.set_line_wrap(True) row.pack_end(labelValue, False, False, 0) return row @@ -38,78 +38,65 @@ def format_generation(gen): raise ValueError("undefined thunderbolt generation") -class DBusProps: - pass +class DBusProperty: + """A read-only property for use with the DBusObject class.""" + def __init__(self, dbus_name): + self.dbus_name = dbus_name + + def __get__(self, instance, owner=None): + if instance is None: + return self + variant = instance._proxy.get_cached_property(self.dbus_name) + return variant.unpack() if variant else None -class DBusProxy: - def __init__(self, name, object_path, interface_name): +class DBusObject: + """A wrapper class around a DBusProxy instance.""" + def __init__(self, bus_name, object_path, interface_name): # Save the dbus info - self.name = name + self.bus_name = bus_name self.object_path = object_path self.interface_name = interface_name - - # Callback for when properties are changed - self.on_property_changed = None - self.on_property_invalidated = None - - # Get the initial properties for this DBus Proxy - var = Gio.DBusConnection.call_sync( - Gio.bus_get_sync(Gio.BusType.SYSTEM), - name, - object_path, - "org.freedesktop.DBus.Properties", - "GetAll", - GLib.Variant("(s)", (interface_name,)), - None, - 0, - -1, - None) - - self.props = DBusProps() - (props,) = var.unpack() - for key, value in props.items(): - setattr(self.props, key, value) - # Main proxy for DBus Object self._proxy = Gio.DBusProxy.new_for_bus_sync( Gio.BusType.SYSTEM, Gio.DBusProxyFlags.NONE, None, - name, + bus_name, object_path, interface_name, None) - + # List of signals that self._proxy is connected to + self._sig_handles = [] + # Callback for when properties are changed + self.on_property_changed = None # Register for future changes to properties - self._proxy.connect("g-properties-changed", self._on_g_properties_changed) + self.signal_connect("g-properties-changed", self._on_g_properties_changed) + + def signal_connect(self, name, callback): + self._sig_handles.append(self._proxy.connect(name, callback)) + + def dispose(self): + for handle in self._sig_handles: + self._proxy.disconnect(handle) + self._sig_handles.clear() + self._proxy = None def _on_g_properties_changed(self, proxy, changed, invalidated): - for key, value in changed.unpack().items(): - setattr(self.props, key, value) - if self.on_property_changed: - self.on_property_changed(key, value) - for key in invalidated: - delattr(self.props, key) - if self.on_property_invalidated: - self.on_property_invalidated(key) + if self.on_property_changed: + self.on_property_changed() -class BoltManagerProxy(DBusProxy): +class BoltManager(DBusObject): + """A DBusObject class for interacting with bolt's Manager interface.""" def __init__(self): # Perform parent initialization - super().__init__( - BOLT_BUS_NAME, - BOLT_OBJECT_PATH, - "org.freedesktop.bolt1.Manager" - ) - + super().__init__(BOLT_BUS_NAME, BOLT_OBJECT_PATH, BOLT_MANAGER_IFACE) # Callbacks self.on_device_added = None self.on_device_removed = None - # Connect to g-signal for event handling - self._proxy.connect('g-signal', self._on_g_signal) + self.signal_connect('g-signal', self._on_g_signal) def _on_g_signal(self, proxy, sender, signal, parameters): if signal == "DeviceAdded" and self.on_device_added: @@ -119,43 +106,52 @@ def _on_g_signal(self, proxy, sender, signal, parameters): (obj_path,) = parameters.unpack() self.on_device_removed(obj_path) - def list_devices(self): + def ListDomains(self): + return self._proxy.ListDomains() + + def ListDevices(self): return self._proxy.ListDevices() - def enroll_device(self, uid): + def EnrollDevice(self, uid): self._proxy.EnrollDevice('(sss)', uid, 'auto', '') - def forget_device(self, uid): + def ForgetDevice(self, uid): self._proxy.ForgetDevice('(s)', uid) -class BoltDeviceProxy(DBusProxy): +class BoltDevice(DBusObject): + """A DBusObject class for interacting with bolt's Device interface.""" + + Vendor = DBusProperty("Vendor") + Name = DBusProperty("Name") + Generation = DBusProperty("Generation") + Type = DBusProperty("Type") + Uid = DBusProperty("Uid") + Status = DBusProperty("Status") + Stored = DBusProperty("Stored") + LinkSpeed = DBusProperty("LinkSpeed") + def __init__(self, obj_path): - super().__init__( - BOLT_BUS_NAME, - obj_path, - "org.freedesktop.bolt1.Device" - ) + super().__init__(BOLT_BUS_NAME, obj_path, BOLT_DEVICE_IFACE) - def authorize(self): + def Authorize(self): self._proxy.Authorize('(s)', 'auto') class BoltSection(SettingsSection): - def __init__(self, bolt_manager, bolt_device): self.bolt_manager = bolt_manager self.bolt_device = bolt_device - self.bolt_device.on_property_changed = lambda k, v: self.refresh() - super().__init__("{0} {1}".format(bolt_device.props.Vendor, bolt_device.props.Name)) - + self.bolt_device.on_property_changed = lambda: self.refresh() + super().__init__("{0} {1}".format(bolt_device.Vendor, bolt_device.Name)) + widget = SettingsWidget() self.status_label = SettingsLabel() widget.pack_start(self.status_label, False, False, 0) self.details_btn = Gtk.ToggleButton(label=_("Details")) self.details_btn.connect("toggled", lambda w: self.details_revealer.set_reveal_child(w.get_active())) self.auth_btn = Gtk.Button(label=_("Authorize")) - self.auth_btn.connect("clicked", lambda w: self.bolt_device.authorize()) + self.auth_btn.connect("clicked", lambda w: self.bolt_device.Authorize()) self.trust_btn = Gtk.Button(label=_("Trust")) self.trust_btn.connect("clicked", self.on_trust_btn_clicked) button_box = Gtk.ButtonBox(orientation=Gtk.Orientation.HORIZONTAL) @@ -165,40 +161,38 @@ def __init__(self, bolt_manager, bolt_device): button_box.set_layout(Gtk.ButtonBoxStyle.EXPAND) widget.pack_end(button_box, False, False, 0) self.add_row(widget) - + list_box = Gtk.ListBox() list_box.set_selection_mode(Gtk.SelectionMode.NONE) list_box.set_header_func(self.update_header) - generation = self.bolt_device.props.Generation + generation = self.bolt_device.Generation list_box.add(build_detail_row(_("Generation"), format_generation(generation))) self.details_bandwidth_label = Gtk.Label(label="-") list_box.add(build_detail_row(_("Bandwidth"), self.details_bandwidth_label)) - dev_type = self.bolt_device.props.Type + dev_type = self.bolt_device.Type list_box.add(build_detail_row(_("Type"), dev_type)) - uid = self.bolt_device.props.Uid + uid = self.bolt_device.Uid list_box.add(build_detail_row("Uid", uid)) self.details_revealer = self.add_reveal_row(list_box) self.refresh() def on_trust_btn_clicked(self, widget): - uid = self.bolt_device.props.Uid - stored = self.bolt_device.props.Stored + uid = self.bolt_device.Uid + stored = self.bolt_device.Stored if stored: - self.bolt_manager.forget_device(uid) + self.bolt_manager.ForgetDevice(uid) else: - self.bolt_manager.enroll_device(uid) + self.bolt_manager.EnrollDevice(uid) def update_header(self, row, before): if before: row.set_header(Gtk.Separator(orientation=Gtk.Orientation.HORIZONTAL)) def refresh(self): - status = self.bolt_device.props.Status - stored = self.bolt_device.props.Stored - link_speed = None - if hasattr(self.bolt_device.props, "LinkSpeed"): - link_speed = self.bolt_device.props.LinkSpeed + status = self.bolt_device.Status + stored = self.bolt_device.Stored + link_speed = self.bolt_device.LinkSpeed # Update the status label text = _("Disconnected") @@ -244,23 +238,16 @@ class Module: def __init__(self, content_box): keywords = _("thunderbolt, usb, docking, station, hub, dock") - sidePage = SidePage("Thunderbolt", "cs-thunderbolt", keywords, content_box, - module=self) + sidePage = SidePage("Thunderbolt", "cs-thunderbolt", keywords, content_box, module=self) self.sidePage = sidePage self.bolt_manager = None self.bolt_devices = dict() - def on_module_selected(self, check_again=False): - # Check if thunderbolt is present - thunderbolt_present = os.path.isdir("/sys/bus/thunderbolt") - # Check if bolt is installed by finding 'boltctl' - bolt_installed = GLib.find_program_in_path("boltctl") - # Check if bolt is available on DBus - boltd_alive = self.test_daemon_alive() + def on_module_selected(self): # Check if we've already been loaded + # This is set by the SidePage class that hosts this module if not self.loaded: - print("Loading Thunderbolt module") self.sidePage.stack = SettingsStack() @@ -300,61 +287,67 @@ def on_module_selected(self, check_again=False): self.sidePage.stack.add_named(page, "settings") page.set_spacing(10) - show_disabled = False - if not thunderbolt_present: - text = _("Thunderbolt or USB4 is not detected on your system.") - self.disabled_retry_button.set_visible(False) - show_disabled = True - elif not bolt_installed: - text = _("The 'bolt' package must be installed to manage Thunderbolt and USB4 devices.") - self.disabled_retry_button.set_visible(True) - self.disabled_retry_button.set_sensitive(True) - show_disabled = True - elif not boltd_alive: - text = _("The boltd service is not running.") + # Check that org.freedesktop.bolt is available on the system + if not self.is_bolt_available(): + text = _("The %s service is missing or not activatable.") % BOLT_BUS_NAME + self.disabled_label.set_markup(f"{text}") self.disabled_retry_button.set_visible(True) self.disabled_retry_button.set_sensitive(True) - show_disabled = True - - if show_disabled: - self.reset() page = "disabled" + GLib.idle_add(self.set_page, page) + return + + # Initialilze the bolt manager + if not self.bolt_manager: + self.bolt_manager = BoltManager() + self.bolt_manager.on_device_added = self.bolt_device_added + self.bolt_manager.on_device_removed = self.bolt_device_removed + + # Check if there are any domains + # If thunderbolt or usb4 is available, there will be 1 domain per controller + # Each domain has a corresponding device with the same uuid + if not self.bolt_manager.ListDomains(): + text = _("Thunderbolt or USB4 is not detected on your system.") self.disabled_label.set_markup(f"{text}") - else: - self.setup() - page = self.page_name() + self.disabled_retry_button.set_visible(False) + page = "disabled" + GLib.idle_add(self.set_page, page) + return - GLib.idle_add(self.set_initial_page, page) + # Setup and display the page + self.setup() + GLib.idle_add(self.set_page, self.page_name()) def disable_retry_on_clicked(self, widget): self.disabled_retry_button.set_sensitive(False) - self.on_module_selected() + GLib.idle_add(self.on_module_selected) - def set_initial_page(self, page): + def set_page(self, page): self.sidePage.stack.set_visible_child_name(page) - def reset(self): - self.bolt_manager = None - for obj_path in list(self.bolt_devices.keys()): - self.bolt_device_removed(obj_path, False) - def setup(self): - if not self.bolt_manager: - self.bolt_manager = BoltManagerProxy() - self.bolt_manager.on_device_added = self.bolt_device_added - self.bolt_manager.on_device_removed = self.bolt_device_removed - for obj_path in self.bolt_manager.list_devices(): - device = BoltDeviceProxy(obj_path) - # Skip the host device - if device.props.Type == "host": + for obj_path in self.bolt_manager.ListDevices(): + params = GLib.Variant("(ss)", (BOLT_DEVICE_IFACE, "Type")) + var = Gio.DBusConnection.call_sync( + Gio.bus_get_sync(Gio.BusType.SYSTEM), + BOLT_BUS_NAME, + obj_path, + "org.freedesktop.DBus.Properties", + "Get", + params, + GLib.VariantType("(v)"), + 0, + -1, + None) + (device_type,) = var.unpack() + if device_type == 'host': continue - # Add the device self.bolt_device_added(obj_path, False) def bolt_device_added(self, obj_path, change_page=True): if obj_path not in self.bolt_devices: # Build the section - device = BoltDeviceProxy(obj_path) + device = BoltDevice(obj_path) section = BoltSection(self.bolt_manager, device) section.show_all() # Add to the page @@ -362,35 +355,39 @@ def bolt_device_added(self, obj_path, change_page=True): page = self.sidePage.stack.get_child_by_name("settings") page.pack_start(section, False, False, 0) if change_page: - self.sidePage.stack.set_visible_child_name(self.page_name()) + GLib.idle_add(self.set_page, self.page_name()) def bolt_device_removed(self, obj_path, change_page=True): if obj_path in self.bolt_devices: section = self.bolt_devices[obj_path] + # Dispose of the bolt device proxy + section.bolt_device.dispose() + section.bolt_device = None + # Destroy the settigs section section.destroy() + # Finally - remove the settings section from the paths dict del self.bolt_devices[obj_path] if change_page: - self.sidePage.stack.set_visible_child_name(self.page_name()) + GLib.idle_add(self.set_page, self.page_name()) def page_name(self): return "settings" if len(self.bolt_devices) > 0 else "empty" - def test_daemon_alive(self): + def is_bolt_available(self): try: - # Ping bolt to see if its available and running - Gio.DBusConnection.call_sync( + var = Gio.DBusConnection.call_sync( Gio.bus_get_sync(Gio.BusType.SYSTEM), - BOLT_BUS_NAME, - BOLT_OBJECT_PATH, - "org.freedesktop.DBus.Peer", - "Ping", - None, + "org.freedesktop.DBus", + "/org/freedesktop/DBus", + "org.freedesktop.DBus", + "ListActivatableNames", None, + GLib.VariantType("(as)"), 0, -1, None) - return True - except GLib.Error as e: - # Bolt isn't installed or service is disabled - pass + (bus_names,) = var.unpack() + return BOLT_BUS_NAME in bus_names + except Exception as e: + print(e) return False