8849220456

Committed 26 Apr 2024 01:40PM UTC coverage: 85.572% (-0.2%) from 85.73%

Build # 8849220456

Build Type

push

github

Committed by

web-flow

Commit Message

chore(screengrab): add comment about auto-delete screenshot file (#621)

Run Details

617 of 803 branches covered (76.84%)

Branch coverage included in aggregate %.

1 of 6 new or added lines in 1 file covered. (16.67%)

19 existing lines in 1 file now uncovered.

2556 of 2905 relevant lines covered (87.99%)

2.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

61.15

/normcap/screengrab/handlers/dbus_portal.py

"""Capture screenshots for all screens using org.freedesktop.portal.Desktop."""

import logging
import random
import re
import sys
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse

from PySide6 import QtCore, QtDBus, QtGui

from normcap.screengrab import system_info
from normcap.screengrab.post_processing import split_full_desktop_to_screens

logger = logging.getLogger(__name__)

install_instructions = ""

# Note on Request Timeout:
#
# Unfortunately, the dbus portal does not return any error message, in case the
# screenshot could not be taken (e.g. missing permission). It just doesn't return any
# response. Therefore, we need to rely on a timeout between the screenshot request and
# its (potential) response to not keep waiting indefinitely.
#
# The value set below is somewhat arbitrary and a trade-off between enabling edge cases
# with high delays (e.g. in high resolution multi monitor setups) and a short delay
# between action and error message, which is desired from a UX perspective.
#
# We started with a 7 seconds timeout, but this turned out to be too low for at least
# one user, therefore it got increased.
#
# ONHOLD: Check in 2024 if the portal was updated to always return a response message.
TIMEOUT_SECONDS = 10


class OrgFreedesktopPortalRequestInterface(QtDBus.QDBusAbstractInterface):
    Response = QtCore.Signal(QtDBus.QDBusMessage)

    def __init__(
        self, path: str, connection: QtDBus.QDBusConnection, parent: QtCore.QObject
    ) -> None:
        super().__init__(
            "org.freedesktop.portal.Desktop",
            path,
            "org.freedesktop.portal.Request",  # type: ignore
            connection,
            parent,
        )


class OrgFreedesktopPortalScreenshot(QtCore.QObject):
    on_response = QtCore.Signal(QtDBus.QDBusMessage)
    on_result = QtCore.Signal(str)
    on_exception = QtCore.Signal(Exception)

    def __init__(
        self,
        parent: Optional[QtCore.QObject] = None,
        interactive: bool = False,
        timeout_sec: int = 15,
    ) -> None:
        super().__init__(parent)
        self.interactive = interactive
        self.timeout_timer = self._get_timeout_timer(timeout_sec)
        self.on_response.connect(self.got_signal)

    def grab_full_desktop(self) -> None:
        bus = QtDBus.QDBusConnection.sessionBus()

        base = bus.baseService()[1:].replace(".", "_")

        random_str = "".join(random.choice("abcdefghi") for _ in range(8))  # noqa: S311
        token = f"normcap_{random_str}"
        object_path = f"/org/freedesktop/portal/desktop/request/{base}/{token}"

        request = OrgFreedesktopPortalRequestInterface(object_path, bus, self)
        request.Response.connect(self.on_response)

        interface = QtDBus.QDBusInterface(
            "org.freedesktop.portal.Desktop",
            "/org/freedesktop/portal/desktop",
            "org.freedesktop.portal.Screenshot",
            bus,
            self,
        )

        message = interface.call(
            "Screenshot", "", {"interactive": False, "handle_token": token}
        )
        logger.debug("DBus request message: %s", str(message))

        if (
            isinstance(message, QtDBus.QDBusMessage)
            and message.arguments()
            and isinstance(message.arguments()[0], QtDBus.QDBusObjectPath)
        ):
            logger.debug("Request accepted")
        else:
            msg = "No object path received from xdg-portal!"
            logger.error(msg)
            self.on_exception.emit(RuntimeError(msg))

    def _get_timeout_timer(self, timeout_sec: int) -> QtCore.QTimer:
        def _timeout_triggered() -> None:
            msg = f"No response from xdg-portal within {timeout_sec}s!"
            logger.error(msg)
            self.on_exception.emit(TimeoutError(msg))

        timeout_timer = QtCore.QTimer()
        timeout_timer.setSingleShot(True)
        timeout_timer.setInterval(timeout_sec * 1000)
        timeout_timer.timeout.connect(_timeout_triggered)
        return timeout_timer

    def got_signal(self, message: QtDBus.QDBusMessage) -> None:
        self.timeout_timer.stop()
        logger.debug("DBus signal message: %s", str(message))

        code, _ = message.arguments()
        all_okay_code = 0
        permission_denied_code = 2

        if code == permission_denied_code:
            msg = f"Permission denied for Screenshot via xdg-portal! Message: {message}"
            logger.error(msg)
            self.on_exception.emit(PermissionError(msg))
            return

        if code != all_okay_code:
            msg = f"Error code {code} received from xdg-portal!"
            logger.error(msg)
            self.on_exception.emit(RuntimeError(msg))
            return

        logger.debug("Parse response")
        # ONHOLD: Extracting DBusArgument as below should work, but it doesn't.
        #         As there doesn't seem to be another way to access the arguments,
        #         we workaround by parsing the URI from string representation.
        # _, arg = message.arguments()
        # QtDBus.QDBusMessage()
        # arg.beginArray()
        # while not arg.atEnd():
        #     arg.beginMap()
        #     while not arg.atEnd():
        #         arg.beginMapEntry()
        #         key = arg.asVariant()
        #         value = arg.asVariant()
        #         arg.endMapEntry()
        #     arg.endMap()
        # arg.endArray()

        reg_uri = r"""
                  \[Variant\(QString\)\:\ \"  # start of URI object
                  (.*)                       # URI itself
                  \"\]\}                     # end of URI object
                  """
        result = re.search(reg_uri, str(message), re.VERBOSE)

        if not result:
            msg = f"Couldn't parse URI from message: {message}"
            logger.error(msg)
            self.on_exception.emit(RuntimeError(message))
            return

        uri = result.group(1)
        self.on_result.emit(uri)


def _synchronized_capture(interactive: bool) -> QtGui.QImage:
    loop = QtCore.QEventLoop()
    result = []
    exceptions = []

    def _signal_triggered(uri: str) -> None:
        result.append(uri)
        loop.exit()

    def _exception_triggered(uri: Exception) -> None:
        exceptions.append(uri)
        loop.exit()

    portal = OrgFreedesktopPortalScreenshot(
        interactive=interactive, timeout_sec=TIMEOUT_SECONDS
    )
    portal.on_result.connect(_signal_triggered)
    portal.on_exception.connect(_exception_triggered)

    portal.timeout_timer.start()
    QtCore.QTimer.singleShot(0, portal.grab_full_desktop)
    loop.exec()

    portal.on_result.disconnect(_signal_triggered)
    portal.on_exception.disconnect(_exception_triggered)

    for error in exceptions:
        raise error

    uri = result[0]
    parsed_uri = urlparse(uri)

    image_path = Path(parsed_uri.path)
    image = QtGui.QImage(image_path)

    # XDG Portal save the image file to the users xdg-pictures directory. To not let
    # them pile up, we try to remove it:
    try:
        image_path.unlink()
    except PermissionError:
        logger.warning("Missing permission to remove screenshot file '%s'!", image_path)

    return image


def is_compatible() -> bool:
    return sys.platform == "linux"


def is_installed() -> bool:
    gnome_version = system_info.get_gnome_version()
    return not gnome_version or gnome_version >= "41"


def capture() -> list[QtGui.QImage]:
    """Capture screenshots for all screens using org.freedesktop.portal.Desktop.

    This methods works gnome-shell >=v41 and wayland.

    In newer xdg-portal implementations, the first request has to be done in
    "interactive" mode, before the application is allowed to query screenshots without
    the dialog window in between.

    As there is no way to query for that permission, we try both:
    1. Try none-interactive mode
    2. If timeout triggers, retry in interactive mode with a helper window
    """
    try:
        image = _synchronized_capture(interactive=False)
    except TimeoutError as exc:
        raise TimeoutError("Timeout when taking screenshot!") from exc
    else:
        return split_full_desktop_to_screens(image)

1	"""Capture screenshots for all screens using org.freedesktop.portal.Desktop."""	3✔
2
3	import logging	3✔
4	import random	3✔
5	import re	3✔
6	import sys	3✔
7	from pathlib import Path	3✔
8	from typing import Optional	3✔
9	from urllib.parse import urlparse	3✔
10
11	from PySide6 import QtCore, QtDBus, QtGui	3✔
12
13	from normcap.screengrab import system_info	3✔
14	from normcap.screengrab.post_processing import split_full_desktop_to_screens	3✔
15
16	logger = logging.getLogger(__name__)	3✔
17
18	install_instructions = ""	3✔
19
20	# Note on Request Timeout:
21	#
22	# Unfortunately, the dbus portal does not return any error message, in case the
23	# screenshot could not be taken (e.g. missing permission). It just doesn't return any
24	# response. Therefore, we need to rely on a timeout between the screenshot request and
25	# its (potential) response to not keep waiting indefinitely.
26	#
27	# The value set below is somewhat arbitrary and a trade-off between enabling edge cases
28	# with high delays (e.g. in high resolution multi monitor setups) and a short delay
29	# between action and error message, which is desired from a UX perspective.
30	#
31	# We started with a 7 seconds timeout, but this turned out to be too low for at least
32	# one user, therefore it got increased.
33	#
34	# ONHOLD: Check in 2024 if the portal was updated to always return a response message.
35	TIMEOUT_SECONDS = 10	3✔
36
37
38	class OrgFreedesktopPortalRequestInterface(QtDBus.QDBusAbstractInterface):	3✔
39	Response = QtCore.Signal(QtDBus.QDBusMessage)	3✔
40
41	def __init__(	3✔
42	self, path: str, connection: QtDBus.QDBusConnection, parent: QtCore.QObject
43	) -> None:
44	super().__init__(	1✔
45	"org.freedesktop.portal.Desktop",
46	path,
47	"org.freedesktop.portal.Request", # type: ignore
48	connection,
49	parent,
50	)
51
52
53	class OrgFreedesktopPortalScreenshot(QtCore.QObject):	3✔
54	on_response = QtCore.Signal(QtDBus.QDBusMessage)	3✔
55	on_result = QtCore.Signal(str)	3✔
56	on_exception = QtCore.Signal(Exception)	3✔
57
58	def __init__(	3✔
59	self,
60	parent: Optional[QtCore.QObject] = None,
61	interactive: bool = False,
62	timeout_sec: int = 15,
63	) -> None:
64	super().__init__(parent)	1✔
65	self.interactive = interactive	1✔
66	self.timeout_timer = self._get_timeout_timer(timeout_sec)	1✔
67	self.on_response.connect(self.got_signal)	1✔
68
69	def grab_full_desktop(self) -> None:	3✔
70	bus = QtDBus.QDBusConnection.sessionBus()	1✔
71
72	base = bus.baseService()[1:].replace(".", "_")	1✔
73
74	random_str = "".join(random.choice("abcdefghi") for _ in range(8)) # noqa: S311	1✔
75	token = f"normcap_{random_str}"	1!
76	object_path = f"/org/freedesktop/portal/desktop/request/{base}/{token}"	1✔
77
78	request = OrgFreedesktopPortalRequestInterface(object_path, bus, self)	1✔
79	request.Response.connect(self.on_response)	1✔
80
81	interface = QtDBus.QDBusInterface(	1✔
82	"org.freedesktop.portal.Desktop",
83	"/org/freedesktop/portal/desktop",
84	"org.freedesktop.portal.Screenshot",
85	bus,
86	self,
87	)
88
89	message = interface.call(	1✔
90	"Screenshot", "", {"interactive": False, "handle_token": token}
91	)
92	logger.debug("DBus request message: %s", str(message))	1✔
93
94	if (	1✔
95	isinstance(message, QtDBus.QDBusMessage)
96	and message.arguments()
97	and isinstance(message.arguments()[0], QtDBus.QDBusObjectPath)
98	):
UNCOV 99	logger.debug("Request accepted")	×
100	else:
101	msg = "No object path received from xdg-portal!"	1!
102	logger.error(msg)	1✔
103	self.on_exception.emit(RuntimeError(msg))	1✔
104
105	def _get_timeout_timer(self, timeout_sec: int) -> QtCore.QTimer:	3✔
106	def _timeout_triggered() -> None:	1✔
107	msg = f"No response from xdg-portal within {timeout_sec}s!"	1✔
108	logger.error(msg)	1✔
109	self.on_exception.emit(TimeoutError(msg))	1✔
110
111	timeout_timer = QtCore.QTimer()	1✔
112	timeout_timer.setSingleShot(True)	1✔
113	timeout_timer.setInterval(timeout_sec * 1000)	1✔
114	timeout_timer.timeout.connect(_timeout_triggered)	1✔
115	return timeout_timer	1✔
116
117	def got_signal(self, message: QtDBus.QDBusMessage) -> None:	3✔
118	self.timeout_timer.stop()	×
UNCOV 119	logger.debug("DBus signal message: %s", str(message))	×
120
121	code, _ = message.arguments()	×
122	all_okay_code = 0	×
UNCOV 123	permission_denied_code = 2	×
124
125	if code == permission_denied_code:	×
126	msg = f"Permission denied for Screenshot via xdg-portal! Message: {message}"	×
127	logger.error(msg)	×
128	self.on_exception.emit(PermissionError(msg))	×
UNCOV 129	return	×
130
131	if code != all_okay_code:	×
132	msg = f"Error code {code} received from xdg-portal!"	×
133	logger.error(msg)	×
134	self.on_exception.emit(RuntimeError(msg))	×
UNCOV 135	return	×
136
UNCOV 137	logger.debug("Parse response")	×
138	# ONHOLD: Extracting DBusArgument as below should work, but it doesn't.
139	# As there doesn't seem to be another way to access the arguments,
140	# we workaround by parsing the URI from string representation.
141	# _, arg = message.arguments()
142	# QtDBus.QDBusMessage()
143	# arg.beginArray()
144	# while not arg.atEnd():
145	# arg.beginMap()
146	# while not arg.atEnd():
147	# arg.beginMapEntry()
148	# key = arg.asVariant()
149	# value = arg.asVariant()
150	# arg.endMapEntry()
151	# arg.endMap()
152	# arg.endArray()
153
UNCOV 154	reg_uri = r"""	×
155	\[Variant\(QString\)\:\ \" # start of URI object
156	(.*) # URI itself
157	\"\]\} # end of URI object
158	"""
UNCOV 159	result = re.search(reg_uri, str(message), re.VERBOSE)	×
160
161	if not result:	×
162	msg = f"Couldn't parse URI from message: {message}"	×
163	logger.error(msg)	×
164	self.on_exception.emit(RuntimeError(message))	×
UNCOV 165	return	×
166
167	uri = result.group(1)	×
UNCOV 168	self.on_result.emit(uri)	×
169
170
171	def _synchronized_capture(interactive: bool) -> QtGui.QImage:	3✔
172	loop = QtCore.QEventLoop()	1✔
173	result = []	1✔
174	exceptions = []	1✔
175
176	def _signal_triggered(uri: str) -> None:	1✔
177	result.append(uri)	×
UNCOV 178	loop.exit()	×
179
180	def _exception_triggered(uri: Exception) -> None:	1✔
181	exceptions.append(uri)	1✔
182	loop.exit()	1✔
183
184	portal = OrgFreedesktopPortalScreenshot(	1✔
185	interactive=interactive, timeout_sec=TIMEOUT_SECONDS
186	)
187	portal.on_result.connect(_signal_triggered)	1✔
188	portal.on_exception.connect(_exception_triggered)	1✔
189
190	portal.timeout_timer.start()	1✔
191	QtCore.QTimer.singleShot(0, portal.grab_full_desktop)	1✔
192	loop.exec()	1✔
193
194	portal.on_result.disconnect(_signal_triggered)	1✔
195	portal.on_exception.disconnect(_exception_triggered)	1✔
196
197	for error in exceptions:	1✔
198	raise error	1!
199
200	uri = result[0]	×
NEW 201	parsed_uri = urlparse(uri)	×
202
NEW 203	image_path = Path(parsed_uri.path)	×
UNCOV 204	image = QtGui.QImage(image_path)	×
205
206	# XDG Portal save the image file to the users xdg-pictures directory. To not let
207	# them pile up, we try to remove it:
UNCOV 208	try:	×
NEW 209	image_path.unlink()	×
NEW 210	except PermissionError:	×
NEW 211	logger.warning("Missing permission to remove screenshot file '%s'!", image_path)	×
212
UNCOV 213	return image	×
214
215
216	def is_compatible() -> bool:	3✔
217	return sys.platform == "linux"	3✔
218
219
220	def is_installed() -> bool:	3✔
221	gnome_version = system_info.get_gnome_version()	3✔
222	return not gnome_version or gnome_version >= "41"	3✔
223
224
225	def capture() -> list[QtGui.QImage]:	3✔
226	"""Capture screenshots for all screens using org.freedesktop.portal.Desktop.
227
228	This methods works gnome-shell >=v41 and wayland.
229
230	In newer xdg-portal implementations, the first request has to be done in
231	"interactive" mode, before the application is allowed to query screenshots without
232	the dialog window in between.
233
234	As there is no way to query for that permission, we try both:
235	1. Try none-interactive mode
236	2. If timeout triggers, retry in interactive mode with a helper window
237	"""
UNCOV 238	try:	×
UNCOV 239	image = _synchronized_capture(interactive=False)	×
UNCOV 240	except TimeoutError as exc:	×
UNCOV 241	raise TimeoutError("Timeout when taking screenshot!") from exc	×
242	else:
UNCOV 243	return split_full_desktop_to_screens(image)	×

dynobo / normcap / 8849220456

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous