Skip to content

Commit

Permalink
dbusutil: Type hinting improvements (#437)
Browse files Browse the repository at this point in the history
* dbusutil: Type hinting improvements

* dbusutil: adjust dbus_progress_message hints

While general DBus argument dicts could have any value,
for the progress message we know the expected valid
types for the message (int or float for progress, int
for count, and bool for the visible setting).

We can type this more explicitly than a generic 'Any'.

* dbusutil: return `bool` in functions
  • Loading branch information
sonic2kk authored Aug 10, 2024
1 parent 140f4d7 commit ff03831
Show file tree
Hide file tree
Showing 2 changed files with 30 additions and 13 deletions.
37 changes: 27 additions & 10 deletions pupgui2/dbusutil.py
Original file line number Diff line number Diff line change
@@ -1,36 +1,53 @@
import os
from typing import Any

from PySide6.QtDBus import QDBusConnection, QDBusMessage

from pupgui2.constants import DBUS_APPLICATION_URI, DBUS_DOWNLOAD_OBJECT_BASEPATH, DBUS_INTERFACES_AND_SIGNALS


def create_and_send_dbus_message(object: str, interface: str, signal_name: str, arguments: list, bus: QDBusConnection = QDBusConnection.sessionBus):
def create_and_send_dbus_message(object: str, interface: str, signal_name: str, arguments: list[Any], bus: QDBusConnection | None = None) -> bool:

"""
Create and send a QDBusMessage over a given bus.
If no bus is given, will default to sessionBus
Returns `True` if the message was sent to DBus successfully, `False` otherwise.
Return Type: bool
"""

if bus is None:
bus = QDBusConnection.sessionBus()

# i.e. /net/davidotek/pupgui2/Update
object_path = os.path.join(DBUS_DOWNLOAD_OBJECT_BASEPATH, object)
object_path: str = os.path.join(DBUS_DOWNLOAD_OBJECT_BASEPATH, object)

message: QDBusMessage = QDBusMessage.createSignal(object_path, interface, signal_name)
message.setArguments(arguments)

# Don't send the message if bus is not valid (i.e. DBus is not running)
if bus.isConnected():
bus.send(message)
return bus.send(message)

return False


def dbus_progress_message(progress: float, count: int = 0, bus: QDBusConnection = QDBusConnection.sessionBus()):
def dbus_progress_message(progress: float, count: int = 0, bus: QDBusConnection | None = None) -> bool:

"""
Create and send download progress (between 0 and 1) information with optional count parameter on a given bus.
If no bus is given, will default to sessionBus.
Returns `True` if the message was sent to DBus successfully, `False` otherwise.
Return Type: bool
"""

arguments = {
if bus is None:
bus = QDBusConnection.sessionBus()

arguments: dict[str, int | float | bool] = {
'progress': progress,
'progress-visible': progress >= 0 and progress < 1,
'count': count,
Expand All @@ -41,17 +58,17 @@ def dbus_progress_message(progress: float, count: int = 0, bus: QDBusConnection
# plus an 'arguments' dict with some extra information
#
# i.e. { 'progress': 0.7, 'progress-visible': True }
message_arguments = [
message_arguments: list[str | dict[str, int | float | bool]] = [
DBUS_APPLICATION_URI,
arguments
]

launcher_entry_update = DBUS_INTERFACES_AND_SIGNALS['LauncherEntryUpdate']
launcher_entry_update: dict[str, str] = DBUS_INTERFACES_AND_SIGNALS['LauncherEntryUpdate']

interface = launcher_entry_update['interface']
signal = launcher_entry_update['signal']
interface: str = launcher_entry_update['interface']
signal: str = launcher_entry_update['signal']
object = 'Update'

create_and_send_dbus_message(object, interface, signal, message_arguments, bus=bus)
return create_and_send_dbus_message(object, interface, signal, message_arguments, bus=bus)


6 changes: 3 additions & 3 deletions pupgui2/pupgui2.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ def __init__(self):
self.msgcb_answer_lock = QMutex()

self.dbus_session_bus = QDBusConnection.sessionBus()
dbus_progress_message(-1, 0) # Reset any previously set download information to be blank
_ = dbus_progress_message(-1, 0) # Reset any previously set download information to be blank

self.load_ui()
self.setup_ui()
Expand Down Expand Up @@ -201,7 +201,7 @@ def send_dbus_download_progress(self, progress: float) -> None:
if progress < 0: # negative progress indicates cancellation/failure/etc
num_downloads = 0

dbus_progress_message(progress_pct, num_downloads, self.dbus_session_bus)
_ = dbus_progress_message(progress_pct, num_downloads, self.dbus_session_bus)

def update_combo_install_location(self, custom_install_dir = None):
self.updating_combo_install_location = True
Expand Down Expand Up @@ -594,5 +594,5 @@ def main():
if IS_FLATPAK and len(os.listdir(STEAM_STL_INSTALL_PATH)) == 0:
subprocess.run(['flatpak-spawn', '--host', 'rm', '-r', STEAM_STL_INSTALL_PATH])

dbus_progress_message(-1, 0) # Reset any previously set download information to be blank
_ = dbus_progress_message(-1, 0) # Reset any previously set download information to be blank
sys.exit(ret)

0 comments on commit ff03831

Please sign in to comment.