refactor(src/services): remove unused properties and methods across service classes

This commit is contained in:
sankalp committed 2026-07-09 18:55:36 +05:45
1 parent f9bef457ca
commit 1403c5adc0
11 files changed
+10 -268

No files matched your search

-36
View File
@@ -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)
+2 -46
View File
@@ -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,
-2
View File
@@ -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
)
+1 -28
View File
@@ -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,
-8
View File
@@ -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."""
+3 -35
View File
@@ -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
-40
View File
@@ -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:
+2 -31
View File
@@ -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()
+2 -12
View File
@@ -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()
-23
View File
@@ -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:
-7
View File
@@ -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)