From 1403c5adc05f1edc9b2bc6c273ac6511454ec3ea Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Thu, 9 Jul 2026 18:55:36 +0545 Subject: [PATCH] refactor(src/services): remove unused properties and methods across service classes --- src/services/battery.py | 36 ---------------------- src/services/bluetooth.py | 48 ++--------------------------- src/services/brightness.py | 2 -- src/services/config.py | 29 +---------------- src/services/custom_notification.py | 8 ----- src/services/globalmenu.py | 38 ++--------------------- src/services/modus.py | 40 ------------------------ src/services/mpris.py | 33 ++------------------ src/services/network.py | 14 ++------- src/services/screencapture.py | 23 -------------- src/services/todo.py | 7 ----- 11 files changed, 10 insertions(+), 268 deletions(-) diff --git a/src/services/battery.py b/src/services/battery.py index 080285a..1d18b61 100644 --- a/src/services/battery.py +++ b/src/services/battery.py @@ -85,18 +85,6 @@ class Battery(Service): def time_to_full(self) -> int: return self.do_get_cached_property("TimeToFull") or 0 - @Property(float, "readable", default_value=0.0) - def energy(self) -> float: - return self.do_get_cached_property("Energy") or 0.0 - - @Property(float, "readable", default_value=0.0) - def energy_full(self) -> float: - return self.do_get_cached_property("EnergyFull") or 0.0 - - @Property(float, "readable", default_value=0.0) - def energy_rate(self) -> float: - return self.do_get_cached_property("EnergyRate") or 0.0 - @Property(float, "readable", default_value=0.0) def temperature(self) -> float: return self.do_get_cached_property("Temperature") or 0.0 @@ -178,30 +166,6 @@ class Battery(Service): def do_handle_power_profile_change(self, *_): self.emit("power_profile_changed") - def do_call_proxy_method( - self, - bus_name, - object_path, - interface_name, - method_name, - parameters=None, - timeout=-1, - ): - if parameters is None: - parameters = GLib.Variant("()", ()) - result = self._bus.call_sync( - bus_name, - object_path, - interface_name, - method_name, - parameters, - None, - Gio.DBusCallFlags.NONE, - timeout, - None, - ) - return result.unpack() - def do_get_cached_property(self, property_name): """Read from the GDBus proxy cache, which DBus keeps up-to-date.""" result = self._proxy.get_cached_property(property_name) diff --git a/src/services/bluetooth.py b/src/services/bluetooth.py index 23097de..eaf802e 100644 --- a/src/services/bluetooth.py +++ b/src/services/bluetooth.py @@ -114,33 +114,10 @@ class BluetoothDevice(Service): def type(self) -> str: return _icon_to_type(str(self._get_prop("Icon") or "")) - @Property(int, "readable") - def battery_level(self) -> int: - return int(self._get_battery_prop("Percentage") or 0) - @Property(float, "readable") def battery_percentage(self) -> float: return float(self._get_battery_prop("Percentage") or 0.0) - @Property(int, "readable") - def rssi(self) -> int: - v = self._get_prop("RSSI") - return int(v) if v is not None else -100 - - @Property(list, "readable") - def uuids(self) -> list: - return list(self._get_prop("UUIDs") or []) - - @Property(int, "readable") - def appearance(self) -> int: - v = self._get_prop("Appearance") - return int(v) if v is not None else 0 - - @Property(int, "readable") - def device_class(self) -> int: - v = self._get_prop("Class") - return int(v) if v is not None else 0 - def __init__( self, bus: Gio.DBusConnection, object_path: str, props: dict, **kwargs ): @@ -269,7 +246,6 @@ class BluetoothDevice(Service): # Removal is done via Adapter1.RemoveDevice — expose a convenience # flag so the adapter can act on it. logger.info(f"[Bluetooth] Remove requested for: {self.address}") - self._removal_requested = True self.emit("changed") def trust(self): @@ -539,9 +515,6 @@ class BluetoothAdapter(Service): self._scan_timeout_id = 0 self.scanning = False - def toggle_power(self): - self.powered = not self.powered - def toggle_scan(self): if self.scanning: self.stop_scan() @@ -578,12 +551,6 @@ class BluetoothClient(Service): @Signal def device_removed(self, address: str) -> None: ... - @Signal - def adapter_added(self, path: str) -> None: ... - - @Signal - def adapter_removed(self, path: str) -> None: ... - @Property(list, "readable") def adapters(self) -> list: return list(self._adapters.values()) @@ -635,9 +602,6 @@ class BluetoothClient(Service): self._adapters: dict[str, BluetoothAdapter] = {} self._bus: Gio.DBusConnection | None = None self._om_proxy: Gio.DBusProxy | None = None - self._iface_added_sub: int = 0 - self._iface_removed_sub: int = 0 - self._bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) self._om_proxy = Gio.DBusProxy.new_sync( self._bus, @@ -649,7 +613,7 @@ class BluetoothClient(Service): None, ) - self._iface_added_sub = self._bus.signal_subscribe( + self._bus.signal_subscribe( BLUEZ_SERVICE, DBUS_OM_IFACE, "InterfacesAdded", @@ -659,7 +623,7 @@ class BluetoothClient(Service): self._on_interfaces_added, None, ) - self._iface_removed_sub = self._bus.signal_subscribe( + self._bus.signal_subscribe( BLUEZ_SERVICE, DBUS_OM_IFACE, "InterfacesRemoved", @@ -723,7 +687,6 @@ class BluetoothClient(Service): lambda *_: self.notify("connected-devices"), ) self._adapters[path] = adapter - self.emit("adapter-added", path) self.notify("adapters") self.emit("changed") @@ -760,7 +723,6 @@ class BluetoothClient(Service): if adapter: logger.info(f"[Bluetooth] Removing adapter: {obj_path}") adapter.close() - self.emit("adapter-removed", obj_path) self.notify("adapters") self.emit("changed") @@ -775,9 +737,6 @@ class BluetoothClient(Service): if a.powered: a.scan() - def toggle_power(self): - self.enabled = not self.enabled - def toggle_scan(self): for a in self._adapters.values(): if a.powered: @@ -789,9 +748,6 @@ class BluetoothClient(Service): return d return None - def get_adapter(self, path: str) -> BluetoothAdapter | None: - return self._adapters.get(path) - def connect_device( self, device: BluetoothDevice, diff --git a/src/services/brightness.py b/src/services/brightness.py index 3444bb6..5f9cf15 100644 --- a/src/services/brightness.py +++ b/src/services/brightness.py @@ -44,7 +44,6 @@ class Brightness(Service): self._last_percent = -1 self._last_raw = -1 self._last_update_time = 0 - self._last_file_mtime = 0 self.backend = self._detect_backend(backend) @@ -69,7 +68,6 @@ class Brightness(Service): else 0 ) - self._last_file_mtime = os.path.getmtime(file_path) self._poll_timer_id = GLib.timeout_add( self.POLL_INTERVAL, self._check_brightness_file ) diff --git a/src/services/config.py b/src/services/config.py index 16c8dab..5f4013c 100644 --- a/src/services/config.py +++ b/src/services/config.py @@ -61,39 +61,12 @@ class ConfigService: """Set a config value (local state only).""" self._config[key] = value - def save(self) -> bool: - """Persist the current config state to disk and trigger reloads.""" - try: - os.makedirs(os.path.dirname(self._config_file), exist_ok=True) - with open(self._config_file, "w") as f: - toml_dump(self._config, f) - - GLib.idle_add(self._reload_config) - return True - except Exception as e: - logger.error(f"[ConfigService] Failed to save config: {e}") - return False - def register_reload_callback( self, callback: Callable[[Dict[str, Any], Dict[str, Any]], None] ) -> None: if callback not in self._reload_callbacks: self._reload_callbacks.append(callback) - def unregister_reload_callback( - self, callback: Callable[[Dict[str, Any], Dict[str, Any]], None] - ) -> None: - if callback in self._reload_callbacks: - self._reload_callbacks.remove(callback) - - def stop(self) -> None: - for monitor in self._monitors: - try: - monitor.cancel() - except Exception as e: - logger.error(f"An error occurred: {e}") - self._monitors.clear() - @staticmethod def _dict_to_toml(d: dict) -> Any: doc = toml_document() @@ -157,7 +130,7 @@ class ConfigService: except Exception as e: logger.error(f"[ConfigService] Failed to monitor {file_path}: {e}") - def _on_file_changed(self, monitor, file, other_file, event_type, file_path: str): + def _on_file_changed(self, monitor, file, _other_file, event_type, file_path: str): # Trigger on various change events to be more robust across editors/OSs valid_events = [ Gio.FileMonitorEvent.CHANGES_DONE_HINT, diff --git a/src/services/custom_notification.py b/src/services/custom_notification.py index fb2414e..9a58891 100644 --- a/src/services/custom_notification.py +++ b/src/services/custom_notification.py @@ -177,14 +177,6 @@ class CachedNotification(Service): self.timestamp = int(time.time()) self._pixbuf_cache: dict[str, GdkPixbuf.Pixbuf | None] = {} - def get_or_cache_pixbuf(self, cache_key: str, loader) -> GdkPixbuf.Pixbuf | None: - if cache_key not in self._pixbuf_cache: - self._pixbuf_cache[cache_key] = loader() - return self._pixbuf_cache[cache_key] - - def remove_from_cache(self): - self.removed_from_cache.emit() - class CachedNotifications(Notifications): """A service to manage the cached notifications.""" diff --git a/src/services/globalmenu.py b/src/services/globalmenu.py index c067e89..819c8bc 100644 --- a/src/services/globalmenu.py +++ b/src/services/globalmenu.py @@ -160,14 +160,6 @@ class GlobalMenuService(Service): def current_menu(self) -> Optional[List[DBusMenuItem]]: return self._current_menu - @Property(str, flags="read-write") - def current_app(self) -> str: - return self._current_app - - @Property(str, flags="read-write") - def current_wm_class(self) -> str: - return self._current_wm_class - def __init__(self): super().__init__() self._current_menu: Optional[List[DBusMenuItem]] = None @@ -302,7 +294,7 @@ class GlobalMenuService(Service): try: self._bus = _get_bus() self._node = Gio.DBusNodeInfo.new_for_xml(REGISTRAR_XML) - self._reg_id = self._bus.register_object( + self._bus.register_object( "/com/canonical/AppMenu/Registrar", self._node.interfaces[0], self._handle_registrar_method, @@ -328,7 +320,7 @@ class GlobalMenuService(Service): connection, sender, object_path, - interface_name, + _interface_name, method_name, parameters, invocation, @@ -475,7 +467,7 @@ class GlobalMenuService(Service): entry.importer.connect_signals( on_layout_updated=on_layout_updated, on_items_updated=on_items_updated, - on_item_activated=lambda item_id, ts: logger.debug( + on_item_activated=lambda item_id, _ts: logger.debug( f"[GlobalMenuService] Item activated: {item_id}" ), ) @@ -1054,30 +1046,6 @@ class GlobalMenuService(Service): return True return False - def about_to_show(self, item_id: int) -> bool: - with self._state_lock: - target_pid = self._current_pid - with self._registry_lock: - svc = self._pid_to_service.get(target_pid) - entry = self._service_registry.get(svc) if svc else None - if entry and entry.importer: - return entry.importer.about_to_show(item_id) - return False - - def refresh_menu_sync(self) -> List[DBusMenuItem]: - with self._state_lock: - target_pid = self._current_pid - with self._registry_lock: - svc = self._pid_to_service.get(target_pid) - entry = self._service_registry.get(svc) if svc else None - if entry and entry.importer: - items = entry.importer.get_layout(force_refresh=True) - with self._state_lock: - if target_pid == self._current_pid: - self._current_menu = items - return items - return [] - _global_menu_svc_instance = None diff --git a/src/services/modus.py b/src/services/modus.py index 6dbe882..4fffc0b 100644 --- a/src/services/modus.py +++ b/src/services/modus.py @@ -64,9 +64,6 @@ class ModusService(Service): @Signal def battery_changed(self, new_battery: str) -> None: ... - @Signal - def dock_apps_changed(self, new_dock_apps: str) -> None: ... - @Signal def dont_disturb_changed(self, value: bool) -> None: ... @@ -222,12 +219,6 @@ class ModusService(Service): self._bluetooth = value self.bluetooth_changed(value) - @dock_apps.setter - def dock_apps(self, value: str): - if value != self._dock_apps: - self._dock_apps = value - self.dock_apps_changed(value) - @dont_disturb.setter def dont_disturb(self, value: bool): if value != self._dont_disturb: @@ -403,25 +394,6 @@ def _on_fullscreen_changed(obj, event): logger.error(f"[ModusService] Error processing fullscreen change: {e}") -def remove_notification(id: int): - get_notification_service().remove_cached_notification(id) - get_modus_service().notification_count_changed(get_notification_service().count) - - -def clear_all_notifications(): - get_notification_service().clear_all_cached_notifications() - get_modus_service().notification_count_changed(get_notification_service().count) - - -def get_cached_notifications(): - return get_notification_service().cached_notifications - - -def get_deserialized_with_ids(): - ns = get_notification_service() - return [(notif._notification, notif.cache_id) for notif in ns.cached_notifications] - - def toggle_dnd(): ns = get_notification_service() ns.toggle_dnd() @@ -447,18 +419,6 @@ def toggle_night_light() -> bool: return False -def set_night_light_temperature(temperature: int) -> bool: - global _night_light_active - try: - _kill_process("hyprsunset") - spawn_detached(["hyprsunset", "-t", str(temperature)]) - _night_light_active = True - return True - except Exception as e: - logger.error(f"[Modus] Failed to set night light temperature: {e}") - return False - - def is_night_light_active() -> bool: global _night_light_active if _night_light_active: diff --git a/src/services/mpris.py b/src/services/mpris.py index df5b5d4..2050bde 100644 --- a/src/services/mpris.py +++ b/src/services/mpris.py @@ -33,30 +33,10 @@ class PlayerService(Service): @Signal def track_position(self, pos: float, dur: float) -> None: ... - @Property(bool, "readable", default_value=False) - def can_go_previous(self) -> bool: - return self._player.get_property("can_go_previous") - - @Property(bool, "readable", default_value=False) - def can_go_next(self) -> bool: - return self._player.get_property("can_go_next") - - @Property(bool, "readable", default_value=False) - def can_pause(self) -> bool: - return self._player.get_property("can_pause") - - @Property(bool, "readable", default_value=False) - def can_play(self) -> bool: - return self._player.get_property("can_play") - @Property(bool, "readable", default_value=False) def can_seek(self) -> bool: return self._player.get_property("can_seek") - @Property(bool, "readable", default_value=False) - def can_control(self) -> bool: - return self._player.get_property("can_control") - @Property(str, "readable", default_value="") def player_name(self) -> str: try: @@ -64,13 +44,6 @@ class PlayerService(Service): except Exception: return "" - @Property(str, "readable", default_value="") - def arturl(self) -> str: - try: - return self._player.props.metadata["mpris:artUrl"] or "" - except Exception: - return "" - @Property(str, "readable", default_value="") def title(self) -> str: try: @@ -143,7 +116,6 @@ class PlayerService(Service): self._current_artwork_hash = "" self._current_artwork_path = "" self._is_cleaning_up = False - self._cached_playback_status = int(self._player.props.playback_status) self._signal_ids = [] self._signal_ids.append( @@ -153,7 +125,6 @@ class PlayerService(Service): self._signal_ids.append(self._player.connect("seeked", self.on_seeked)) self.status = self._player.props.playback_status - self._last_polled_status = self.playback_status self._pos_polling = False # guard: Fabricator.start() blindly creates a new # GLib timer every call — this flag prevents stacking self.pos_fabricator = Fabricator( @@ -435,10 +406,10 @@ class PlayerManager(Service): except Exception as e: logger.error(f"Failed to create player {name_str}: {e}") - def _on_name_appeared(self, sender, name, manager): + def _on_name_appeared(self, sender, name, _manager): self._create_player(name) - def _on_player_vanished(self, sender, player, manager): + def _on_player_vanished(self, sender, player, _manager): name = player.props.player_name if name in self._services: self._services[name].cleanup() diff --git a/src/services/network.py b/src/services/network.py index d4a32b3..a0c5ac8 100644 --- a/src/services/network.py +++ b/src/services/network.py @@ -18,9 +18,6 @@ class Wifi(Service): @Signal def changed(self) -> None: ... - @Signal - def enabled_changed(self) -> bool: ... - def __init__(self, client: NM.Client, device: NM.DeviceWifi, **kwargs): self._client: NM.Client = client self._device: NM.DeviceWifi = device @@ -213,9 +210,6 @@ class Ethernet(Service): @Signal def changed(self) -> None: ... - @Signal - def enabled(self) -> bool: ... - @Property(int, "readable") def speed(self) -> int: return self._device.get_speed() @@ -369,7 +363,7 @@ class NetworkClient(Service): device = self.wifi_device._device handler_id: list[int] = [] - def _on_state_changed(dev, new_state, old_state, reason): + def _on_state_changed(_dev, new_state, _old_state, reason): if not callback: return state = NM.DeviceState(new_state) @@ -410,7 +404,7 @@ class NetworkClient(Service): device = self.wifi_device._device handler_id: list[int] = [] - def _on_state_changed(dev, new_state, old_state, reason): + def _on_state_changed(_dev, new_state, _old_state, reason): state = NM.DeviceState(new_state) if state == NM.DeviceState.ACTIVATED: _cleanup() @@ -439,7 +433,3 @@ class NetworkClient(Service): f"nmcli device wifi connect {bssid} password {password!r}", lambda *args: logger.debug(f"connect_with_password result: {args}"), ) - - @Property(str, "readable") - def primary_device(self) -> Literal["wifi", "wired"] | None: - return self._get_primary_device() diff --git a/src/services/screencapture.py b/src/services/screencapture.py index 9f2fd89..1caf2fa 100644 --- a/src/services/screencapture.py +++ b/src/services/screencapture.py @@ -148,29 +148,6 @@ class ScreenCapture(Service): proc.communicate_utf8_async(None, None, do_callback) self.recording_stopped(file_path) - def check_wf_recorder(self): - if not is_app_running("wf-recorder"): - return False - - self.stop_recording() - return True - - def record_video(self, output_file, *args): - cmd = [ - "wf-recorder", - *args, - "-f", - str(output_file), - "-c", - "libvpx-vp9", - "--pixel-format", - "yuv420p", - "-F", - "eq=brightness=0.12:contrast=1.1", - "--no-audio", - ] - return run_command(cmd) - def _get_active_monitor(self): """Get the name of the currently focused monitor.""" try: diff --git a/src/services/todo.py b/src/services/todo.py index b7d259e..54e47c9 100644 --- a/src/services/todo.py +++ b/src/services/todo.py @@ -127,13 +127,6 @@ class TodoService(Service): return True return False - def get_todo(self, todo_id: str) -> dict | None: - """Get a specific todo by ID""" - for todo in self._todos: - if todo["id"] == todo_id: - return todo.copy() - return None - def clear_completed(self): """Remove all completed todos""" initial_count = len(self._todos)