KDE Job progress via python-dbus

kde_dbus_job.py

#!/usr/bin/env python
# coding: utf-8
"""
Python wrapper around the KDE JobViewServer DBus interface
"""

import dbus
import enum
import time
import typing


class Unit(enum.Enum):
    Bytes = enum.auto()
    Files = enum.auto()
    Directories = enum.auto()


class KDEDBusJob:

    KUISERVER = 'org.kde.kuiserver'
    JOB_INTERFACE = 'org.kde.JobViewV2'
    JOB_VIEW_INTERFACE = 'org.kde.JobViewServer'
    JOB_VIEW_PATH = '/JobViewServer'

    def __init__(self, name, icon_name):
        self.job = None
        self.count = 0
        self.bus = dbus.SessionBus()
        self.job_view_server = self.get_dbus_interface(
            self.JOB_VIEW_PATH, self.JOB_VIEW_INTERFACE)

        self.name = name
        self.icon_name = icon_name
        self.unit = ""

    def __del__(self):
        self.finish("Terminated.")

    def get_dbus_interface(self, path, interface):
        obj = self.bus.get_object(self.KUISERVER, path)
        return dbus.Interface(obj, dbus_interface=interface)

    def start(self, info_msg):
        "Starts the job, info_msg is shown as title in the job progress dialog"
        if self.job is not None:
            return

        # # KJob::Capability 0: None, 1: Killable, 2: Suspendable
        request_path = self.job_view_server.requestView(
            self.name, self.icon_name, 3)
        self.job = self.get_dbus_interface(request_path, self.JOB_INTERFACE)
        self.job.setInfoMessage(info_msg)

    def finish(self, msg="", dest_url=None):
        "Finishes the job with msg. If dest_url is given an open dialog is shown"
        if self.job is not None:
            self.job.clearDescriptionField(0)
            self.job.clearDescriptionField(1)
            if dest_url is not None:
                self.job.setDestUrl(dest_url)
                self.job.setDescriptionField(1, "Copied to: ", dest_url)
            self.job.terminate("Done")
            self.job = None

    def update_amount(self, count: str, unit: typing.Union[str, Unit]):
        "Update the current amount. Note: You can have one amount value per units"
        if isinstance(unit, Unit):
            unit = unit.name
        if self.job is not None:
            self.job.setProcessedAmount(count, unit)

    def update_description(self, field: int, description: str, value: str):
        if self.job is not None:
            self.job.setDescriptionField(field, description, value)

    def update_description_0(self, description: str, value: str):
        self.update_description(0, description, value)

    def update_description_1(self, description: str, value: str):
        self.update_description(1, description, value)

    def update_percent(self, value: int):
        if self.job is not None:
            self.job.setPercent(value)

    def update_speed(self, value: int):
        if self.job is not None:
            self.job.setSpeed(value)

    def update_total_amount(self, count: int, unit: typing.Union[str, Unit]):
        """Sets the total amount.

        Note: The amount for each unit is shown, so that you can have multiple
        amount values.
        """
        if isinstance(unit, Unit):
            unit = unit.name
        if self.job is not None:
            self.job.setTotalAmount(count, unit)


def test():
    cnt = 4
    sub_steps = 2

    dialog = KDEDBusJob("Test dialog", "test")
    dialog.start("Counting...")
    dialog.update_total_amount(cnt, Unit.Files)
    for ii in range(cnt):
        dialog.update_amount(ii, Unit.Files)
        dialog.update_description(0, "ItemA", "A{}".format(ii))
        dialog.update_description(1, "ItemB", "B{}".format(ii))
        dialog.update_total_amount(sub_steps, "steps")
        for s in range(sub_steps):
            percent = s * 100 // sub_steps
            dialog.update_speed(1 / sub_steps)
            dialog.update_percent(percent)
            dialog.update_amount(s, "steps")
            time.sleep(1)
    dialog.finish("Done.", dest_url='/tmp')


if __name__ == '__main__':
    test()

Without some kind of license declaration, the terms under which this may be used are iffy.

As the person whose sample this code was adapted from, in case this counts as a derivative work, I hereby give syntion full permission to apply whatever license they want.