3886830687

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/qtile/qtile/commit/<a class=hub.com/qtile/qtile/commit/<a class="double-link" href="https://git"><a class=hub.com/qtile/qtile/commit/<a class="double-link" href="https://git"><a class=hub.com/qtile/qtile/commit/<a class="double-link" href="https://git"><a class=hub.com/qtile/qtile/commit/7e3a975cc93b2e1c0e78097e5d65a2313283d290">7e3a975cc">&lt;a href=&quot;https://github.com/qtile/qtile/commit/</a><a class="double-link" href="https://github.com/qtile/qtile/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/qtile/qtile/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/qtile/qtile/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/qtile/qtile/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/qtile/qtile/commit/7e3a975cc93b2e1c0e78097e5d65a2313283d290">7e3a975cc</a><a href="https://github.com/qtile/qtile/commit/7e3a975cc93b2e1c0e78097e5d65a2313283d290">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/qtile/qtile/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/qtile/qtile/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/qtile/qtile/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/qtile/qtile/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/qtile/qtile/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/qtile/qtile/commit/7e3a975cc93b2e1c0e78097e5d65a2313283d290&quot;&gt;7e3a975cc&lt;/a&gt;&lt;a href=&quot;https://github.com/qtile/qtile/commit/7e3a975cc93b2e1c0e78097e5d65a2313283d290&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/qtile/qtile/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/qtile/qtile/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp... (continued)

Pull Request Pull Request #3543: Remove dbus match rules on shutdown/restart/reload

Run Details

23 of 33 new or added lines in 3 files covered. (69.7%)

13659 of 18459 relevant lines covered (74.0%)

2.21 hits per line

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

65.91

/libqtile/widget/mpris2widget.py

# Copyright (c) 2014 Sebastian Kricner
# Copyright (c) 2014 Sean Vig
# Copyright (c) 2014 Adi Sieker
# Copyright (c) 2014 Tycho Andersen
# Copyright (c) 2020 elParaguayo
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from __future__ import annotations

import asyncio
import re
import string
from typing import TYPE_CHECKING

from dbus_next import Message, Variant
from dbus_next.aio import MessageBus
from dbus_next.constants import MessageType

from libqtile.command.base import expose_command
from libqtile.log_utils import logger
from libqtile.utils import _send_dbus_message, add_signal_receiver
from libqtile.widget import base

if TYPE_CHECKING:
    from typing import Any

MPRIS_PATH = "/org/mpris/MediaPlayer2"
MPRIS_OBJECT = "org.mpris.MediaPlayer2"
MPRIS_PLAYER = "org.mpris.MediaPlayer2.Player"
PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"
MPRIS_REGEX = re.compile(r"(\{(.*?):(.*?)(:.*?)?\})")


class Mpris2Formatter(string.Formatter):
    """
    Custom string formatter for MPRIS2 metadata.

    Keys have a colon (e.g. "xesam:title") which causes issues with python's string
    formatting as the colon splits the identifier from the format specification.

    This formatter handles this issue by changing the first colon to an underscore and
    then formatting the incoming kwargs to match.

    Additionally, a default value is returned when an identifier is not provided by the
    kwarg data.
    """

    def __init__(self, default=""):
        string.Formatter.__init__(self)
        self._default = default

    def get_value(self, key, args, kwargs):
        """
        Replaces colon in kwarg keys with an underscore before getting value.

        Missing identifiers are replaced with the default value.
        """
        kwargs = {k.replace(":", "_"): v for k, v in kwargs.items()}
        try:
            return string.Formatter.get_value(self, key, args, kwargs)
        except (IndexError, KeyError):
            return self._default

    def parse(self, format_string):
        """
        Replaces first colon in format string with an underscore.

        This will cause issues if any identifier is provided that does not
        contain a colon. This should not happen according to the MPRIS2
        specification!
        """
        format_string = MPRIS_REGEX.sub(r"{\2_\3\4}", format_string)
        return string.Formatter.parse(self, format_string)


class Mpris2(base._TextBox):
    """An MPRIS 2 widget

    A widget which displays the current track/artist of your favorite MPRIS
    player. This widget scrolls the text if neccessary and information that
    is displayed is configurable.

    The widget relies on players broadcasting signals when the metadata or playback
    status changes. If you are getting inconsistent results then you can enable background
    polling of the player by setting the `poll_interval` parameter. This is disabled by
    default.

    Basic mouse controls are also available: button 1 = play/pause,
    scroll up = next track, scroll down = previous track.

    Widget requirements: dbus-next_.

    .. _dbus-next: https://pypi.org/project/dbus-next/
    """

    defaults = [
        ("name", "audacious", "Name of the MPRIS widget."),
        (
            "objname",
            None,
            "DBUS MPRIS 2 compatible player identifier"
            "- Find it out with dbus-monitor - "
            "Also see: http://specifications.freedesktop.org/"
            "mpris-spec/latest/#Bus-Name-Policy. "
            "``None`` will listen for notifications from all MPRIS2 compatible players.",
        ),
        (
            "format",
            "{xesam:title} - {xesam:album} - {xesam:artist}",
            "Format string for displaying metadata. "
            "See http://www.freedesktop.org/wiki/Specifications/mpris-spec/metadata/#index5h3 "
            "for available values",
        ),
        ("separator", ", ", "Separator for metadata fields that are a list."),
        (
            "display_metadata",
            ["xesam:title", "xesam:album", "xesam:artist"],
            "(Deprecated) Which metadata identifiers to display. ",
        ),
        ("scroll", True, "Whether text should scroll."),
        ("playing_text", "{track}", "Text to show when playing"),
        ("paused_text", "Paused: {track}", "Text to show when paused"),
        ("stopped_text", "", "Text to show when stopped"),
        (
            "stop_pause_text",
            None,
            "(Deprecated) Optional text to display when in the stopped/paused state",
        ),
        (
            "no_metadata_text",
            "No metadata for current track",
            "Text to show when track has no metadata",
        ),
        (
            "poll_interval",
            0,
            "Periodic background polling interval of player (0 to disable polling).",
        ),
    ]

    def __init__(self, **config):
        base._TextBox.__init__(self, "", **config)
        self.add_defaults(Mpris2.defaults)
        self.is_playing = False
        self.count = 0
        self.displaytext = ""
        self.track_info = ""
        self.status = "{track}"
        self.add_callbacks(
            {
                "Button1": self.play_pause,
                "Button4": self.next,
                "Button5": self.previous,
            }
        )
        paused = ""
        stopped = ""
        if "stop_pause_text" in config:
            logger.warning(
                "The use of 'stop_pause_text' is deprecated. Please use 'paused_text' and 'stopped_text' instead."
            )
            if "paused_text" not in config:
                paused = self.stop_pause_text

            if "stopped_text" not in config:
                stopped = self.stop_pause_text

        if "display_metadata" in config:
            logger.warning(
                "The use of `display_metadata is deprecated. Please use `format` instead."
            )
            self.format = " - ".join(config["display_metadata"])

        self._formatter = Mpris2Formatter()

        self.prefixes = {
            "Playing": self.playing_text,
            "Paused": paused or self.paused_text,
            "Stopped": stopped or self.stopped_text,
        }

        self._current_player: str | None = None
        self.player_names: dict[str, str] = {}
        self._background_poll: asyncio.TimerHandle | None = None
        self.bus: MessageBus | None = None

    @property
    def player(self) -> str:
        if self._current_player is None:
            return "None"
        else:
            return self.player_names.get(self._current_player, "Unknown")

    async def _config_async(self):
        # These two listeners create separate bus connections. Each connection only has one
        # callback so we don't need any logic to identify the message and the appropriate
        # handler in this code.

        # Set up a listener for NameOwner changes so we can remove players when they close
        await add_signal_receiver(
            self._name_owner_changed,
            session_bus=True,
            signal_name="NameOwnerChanged",
            dbus_interface="org.freedesktop.DBus",
        )

        # Listen out for signals from any Mpris2 compatible player
        subscribe = await add_signal_receiver(
            self.message,
            session_bus=True,
            signal_name="PropertiesChanged",
            bus_name=self.objname,
            path="/org/mpris/MediaPlayer2",
            dbus_interface="org.freedesktop.DBus.Properties",
        )

        if not subscribe:
            logger.warning("Unable to add signal receiver for Mpris2 players")

        # If the user has specified a player to be monitored, we can poll it now.
        if self.objname is not None:
            await self._check_player()

    def _name_owner_changed(self, message):
        # We need to track when an interface has been removed from the bus
        # We use the NameOwnerChanged signal and check if the new owner is
        # empty.
        name, _, new_owner = message.body

        # Check if the current player has closed
        if new_owner == "" and name == self._current_player:
            self._current_player = None
            self.update("")

            # Cancel any scheduled background poll
            self._set_background_poll(False)

    def message(self, message):
        if message.message_type != MessageType.SIGNAL:
            return

        asyncio.create_task(self.process_message(message))

    async def process_message(self, message):
        current_player = message.sender

        if current_player not in self.player_names:
            self.player_names[current_player] = await self.get_player_name(current_player)

        self._current_player = current_player

        self.parse_message(*message.body)

    async def _send_message(self, destination, interface, path, member, signature, body):
        bus, message = await _send_dbus_message(
            session_bus=True,
            message_type=MessageType.METHOD_CALL,
            destination=destination,
            interface=interface,
            path=path,
            member=member,
            signature=signature,
            body=body,
            bus=self.bus,
        )

        # We should reuse the same bus connection for repeated calls.
        if self.bus is None:
            self.bus = bus

        return message

    async def _check_player(self):
        """Check for player at startup and retrieve metadata."""
        if not (self.objname or self._current_player):
            return

        message = await self._send_message(
            self.objname if self.objname else self._current_player,
            PROPERTIES_INTERFACE,
            MPRIS_PATH,
            "GetAll",
            "s",
            [MPRIS_PLAYER],
        )

        # If we get an error here it will be because the player object doesn't exist
        if message.message_type != MessageType.METHOD_RETURN:
            self._current_player = None
            self.update("")
            return

        if message.body:
            self._current_player = message.sender
            self.parse_message(self.objname, message.body[0], [])

    def _set_background_poll(self, poll=True):
        if self._background_poll is not None:
            self._background_poll.cancel()

        if poll:
            self._background_poll = self.timeout_add(self.poll_interval, self._check_player)

    async def get_player_name(self, player):
        message = await self._send_message(
            player,
            PROPERTIES_INTERFACE,
            MPRIS_PATH,
            "Get",
            "ss",
            [MPRIS_OBJECT, "Identity"],
        )

        if message.message_type != MessageType.METHOD_RETURN:
            logger.warning("Could not retrieve identity of player on %s.", player)
            return ""

        return message.body[0].value

    def parse_message(
        self,
        _interface_name: str,
        changed_properties: dict[str, Any],
        _invalidated_properties: list[str],
    ) -> None:
        """
        http://specifications.freedesktop.org/mpris-spec/latest/Track_List_Interface.html#Mapping:Metadata_Map
        """
        if not self.configured:
            return

        if "Metadata" not in changed_properties and "PlaybackStatus" not in changed_properties:
            return

        self.displaytext = ""

        metadata = changed_properties.get("Metadata")
        if metadata:
            self.track_info = self.get_track_info(metadata.value)

        playbackstatus = getattr(changed_properties.get("PlaybackStatus"), "value", None)
        if playbackstatus:
            self.is_playing = playbackstatus == "Playing"
            self.status = self.prefixes.get(playbackstatus, "{track}")

        if not self.track_info:
            self.track_info = self.no_metadata_text

        self.displaytext = self.status.format(track=self.track_info)

        if self.text != self.displaytext:
            self.update(self.displaytext)

        if self.poll_interval:
            self._set_background_poll()

    def get_track_info(self, metadata: dict[str, Variant]) -> str:
        self.metadata = {}
        for key in metadata:
            new_key = key
            val = getattr(metadata.get(key), "value", None)
            if isinstance(val, str):
                self.metadata[new_key] = val
            elif isinstance(val, list):
                self.metadata[new_key] = self.separator.join(
                    (y for y in val if isinstance(y, str))
                )

        return self._formatter.format(self.format, **self.metadata).replace("\n", "")

    def _player_cmd(self, cmd: str) -> None:
        if self._current_player is None:
            return

        task = asyncio.create_task(self._send_player_cmd(cmd))
        task.add_done_callback(self._task_callback)

    async def _send_player_cmd(self, cmd: str) -> Message | None:
        message = await self._send_message(
            self._current_player,
            MPRIS_PLAYER,
            MPRIS_PATH,
            cmd,
            "",
            [],
        )

        return message

    def _task_callback(self, task: asyncio.Task) -> None:
        message = task.result()

        # This happens if we can't connect to dbus. Logger call is made
        # elsewhere so we don't need to do any more here.
        if message is None:
            return

        if message.message_type != MessageType.METHOD_RETURN:
            logger.warning("Unable to send command to player.")

    @expose_command()
    def play_pause(self) -> None:
        """Toggle the playback status."""
        self._player_cmd("PlayPause")

    @expose_command()
    def next(self) -> None:
        """Play the next track."""
        self._player_cmd("Next")

    @expose_command()
    def previous(self) -> None:
        """Play the previous track."""
        self._player_cmd("Previous")

    @expose_command()
    def stop(self) -> None:
        """Stop playback."""
        self._player_cmd("Stop")

    @expose_command()
    def info(self):
        """What's the current state of the widget?"""
        d = base._TextBox.info(self)
        d.update(dict(isplaying=self.is_playing, player=self.player))
        return d

1	# Copyright (c) 2014 Sebastian Kricner
2	# Copyright (c) 2014 Sean Vig
3	# Copyright (c) 2014 Adi Sieker
4	# Copyright (c) 2014 Tycho Andersen
5	# Copyright (c) 2020 elParaguayo
6	#
7	# Permission is hereby granted, free of charge, to any person obtaining a copy
8	# of this software and associated documentation files (the "Software"), to deal
9	# in the Software without restriction, including without limitation the rights
10	# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11	# copies of the Software, and to permit persons to whom the Software is
12	# furnished to do so, subject to the following conditions:
13	#
14	# The above copyright notice and this permission notice shall be included in
15	# all copies or substantial portions of the Software.
16	#
17	# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18	# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19	# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20	# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21	# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22	# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23	# SOFTWARE.
24	from __future__ import annotations	3✔
25
26	import asyncio	3✔
27	import re	3✔
28	import string	3✔
29	from typing import TYPE_CHECKING	3✔
30
31	from dbus_next import Message, Variant	3✔
32	from dbus_next.aio import MessageBus	3✔
33	from dbus_next.constants import MessageType	3✔
34
35	from libqtile.command.base import expose_command	3✔
36	from libqtile.log_utils import logger	3✔
37	from libqtile.utils import _send_dbus_message, add_signal_receiver	3✔
38	from libqtile.widget import base	3✔
39
40	if TYPE_CHECKING:	3✔
41	from typing import Any	×
42
43	MPRIS_PATH = "/org/mpris/MediaPlayer2"	3✔
44	MPRIS_OBJECT = "org.mpris.MediaPlayer2"	3✔
45	MPRIS_PLAYER = "org.mpris.MediaPlayer2.Player"	3✔
46	PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"	3✔
47	MPRIS_REGEX = re.compile(r"(\{(.?):(.?)(:.*?)?\})")	3✔
48
49
50	class Mpris2Formatter(string.Formatter):	3✔
51	"""
52	Custom string formatter for MPRIS2 metadata.
53
54	Keys have a colon (e.g. "xesam:title") which causes issues with python's string
55	formatting as the colon splits the identifier from the format specification.
56
57	This formatter handles this issue by changing the first colon to an underscore and
58	then formatting the incoming kwargs to match.
59
60	Additionally, a default value is returned when an identifier is not provided by the
61	kwarg data.
62	"""
63
64	def __init__(self, default=""):	3✔
65	string.Formatter.__init__(self)	3✔
66	self._default = default	3✔
67
68	def get_value(self, key, args, kwargs):	3✔
69	"""
70	Replaces colon in kwarg keys with an underscore before getting value.
71
72	Missing identifiers are replaced with the default value.
73	"""
74	kwargs = {k.replace(":", "_"): v for k, v in kwargs.items()}	3✔
75	try:	3✔
76	return string.Formatter.get_value(self, key, args, kwargs)	3✔
77	except (IndexError, KeyError):	×
78	return self._default	×
79
80	def parse(self, format_string):	3✔
81	"""
82	Replaces first colon in format string with an underscore.
83
84	This will cause issues if any identifier is provided that does not
85	contain a colon. This should not happen according to the MPRIS2
86	specification!
87	"""
88	format_string = MPRIS_REGEX.sub(r"{\2_\3\4}", format_string)	3✔
89	return string.Formatter.parse(self, format_string)	3✔
90
91
92	class Mpris2(base._TextBox):	3✔
93	"""An MPRIS 2 widget
94
95	A widget which displays the current track/artist of your favorite MPRIS
96	player. This widget scrolls the text if neccessary and information that
97	is displayed is configurable.
98
99	The widget relies on players broadcasting signals when the metadata or playback
100	status changes. If you are getting inconsistent results then you can enable background
101	polling of the player by setting the `poll_interval` parameter. This is disabled by
102	default.
103
104	Basic mouse controls are also available: button 1 = play/pause,
105	scroll up = next track, scroll down = previous track.
106
107	Widget requirements: dbus-next_.
108
109	.. _dbus-next: https://pypi.org/project/dbus-next/
110	"""
111
112	defaults = [	3✔
113	("name", "audacious", "Name of the MPRIS widget."),
114	(
115	"objname",
116	None,
117	"DBUS MPRIS 2 compatible player identifier"
118	"- Find it out with dbus-monitor - "
119	"Also see: http://specifications.freedesktop.org/"
120	"mpris-spec/latest/#Bus-Name-Policy. "
121	"``None`` will listen for notifications from all MPRIS2 compatible players.",
122	),
123	(
124	"format",
125	"{xesam:title} - {xesam:album} - {xesam:artist}",
126	"Format string for displaying metadata. "
127	"See http://www.freedesktop.org/wiki/Specifications/mpris-spec/metadata/#index5h3 "
128	"for available values",
129	),
130	("separator", ", ", "Separator for metadata fields that are a list."),
131	(
132	"display_metadata",
133	["xesam:title", "xesam:album", "xesam:artist"],
134	"(Deprecated) Which metadata identifiers to display. ",
135	),
136	("scroll", True, "Whether text should scroll."),
137	("playing_text", "{track}", "Text to show when playing"),
138	("paused_text", "Paused: {track}", "Text to show when paused"),
139	("stopped_text", "", "Text to show when stopped"),
140	(
141	"stop_pause_text",
142	None,
143	"(Deprecated) Optional text to display when in the stopped/paused state",
144	),
145	(
146	"no_metadata_text",
147	"No metadata for current track",
148	"Text to show when track has no metadata",
149	),
150	(
151	"poll_interval",
152	0,
153	"Periodic background polling interval of player (0 to disable polling).",
154	),
155	]
156
157	def __init__(self, **config):	3✔
158	base._TextBox.__init__(self, "", **config)	3✔
159	self.add_defaults(Mpris2.defaults)	3✔
160	self.is_playing = False	3✔
161	self.count = 0	3✔
162	self.displaytext = ""	3✔
163	self.track_info = ""	3✔
164	self.status = "{track}"	3✔
165	self.add_callbacks(	3✔
166	{
167	"Button1": self.play_pause,
168	"Button4": self.next,
169	"Button5": self.previous,
170	}
171	)
172	paused = ""	3✔
173	stopped = ""	3✔
174	if "stop_pause_text" in config:	3✔
175	logger.warning(	3✔
176	"The use of 'stop_pause_text' is deprecated. Please use 'paused_text' and 'stopped_text' instead."
177	)
178	if "paused_text" not in config:	3✔
179	paused = self.stop_pause_text	3✔
180
181	if "stopped_text" not in config:	3✔
182	stopped = self.stop_pause_text	3✔
183
184	if "display_metadata" in config:	3✔
185	logger.warning(	×
186	"The use of `display_metadata is deprecated. Please use `format` instead."
187	)
188	self.format = " - ".join(config["display_metadata"])	×
189
190	self._formatter = Mpris2Formatter()	3✔
191
192	self.prefixes = {	3✔
193	"Playing": self.playing_text,
194	"Paused": paused or self.paused_text,
195	"Stopped": stopped or self.stopped_text,
196	}
197
198	self._current_player: str \| None = None	3✔
199	self.player_names: dict[str, str] = {}	3✔
200	self._background_poll: asyncio.TimerHandle \| None = None	3✔
201	self.bus: MessageBus \| None = None	3✔
202
203	@property	3✔
204	def player(self) -> str:	3✔
205	if self._current_player is None:	3✔
206	return "None"	3✔
207	else:
208	return self.player_names.get(self._current_player, "Unknown")	×
209
210	async def _config_async(self):	3✔
211	# These two listeners create separate bus connections. Each connection only has one
212	# callback so we don't need any logic to identify the message and the appropriate
213	# handler in this code.
214
215	# Set up a listener for NameOwner changes so we can remove players when they close
216	await add_signal_receiver(	3✔
217	self._name_owner_changed,
218	session_bus=True,
219	signal_name="NameOwnerChanged",
220	dbus_interface="org.freedesktop.DBus",
221	)
222
223	# Listen out for signals from any Mpris2 compatible player
224	subscribe = await add_signal_receiver(	3✔
225	self.message,
226	session_bus=True,
227	signal_name="PropertiesChanged",
228	bus_name=self.objname,
229	path="/org/mpris/MediaPlayer2",
230	dbus_interface="org.freedesktop.DBus.Properties",
231	)
232
233	if not subscribe:	3✔
234	logger.warning("Unable to add signal receiver for Mpris2 players")	3✔
235
236	# If the user has specified a player to be monitored, we can poll it now.
237	if self.objname is not None:	3✔
238	await self._check_player()	×
239
240	def _name_owner_changed(self, message):	3✔
241	# We need to track when an interface has been removed from the bus
242	# We use the NameOwnerChanged signal and check if the new owner is
243	# empty.
244	name, _, new_owner = message.body	×
245
246	# Check if the current player has closed
247	if new_owner == "" and name == self._current_player:	×
248	self._current_player = None	×
249	self.update("")	×
250
251	# Cancel any scheduled background poll
252	self._set_background_poll(False)	×
253
254	def message(self, message):	3✔
255	if message.message_type != MessageType.SIGNAL:	×
256	return	×
257
258	asyncio.create_task(self.process_message(message))	×
259
260	async def process_message(self, message):	3✔
261	current_player = message.sender	×
262
263	if current_player not in self.player_names:	×
264	self.player_names[current_player] = await self.get_player_name(current_player)	×
265
266	self._current_player = current_player	×
267
268	self.parse_message(*message.body)	×
269
270	async def _send_message(self, destination, interface, path, member, signature, body):	3✔
NEW 271	bus, message = await _send_dbus_message(	×
272	session_bus=True,
273	message_type=MessageType.METHOD_CALL,
274	destination=destination,
275	interface=interface,
276	path=path,
277	member=member,
278	signature=signature,
279	body=body,
280	bus=self.bus,
281	)
282
283	# We should reuse the same bus connection for repeated calls.
NEW 284	if self.bus is None:	×
NEW 285	self.bus = bus	×
286
NEW 287	return message	×
288
289	async def _check_player(self):	3✔
290	"""Check for player at startup and retrieve metadata."""
291	if not (self.objname or self._current_player):	×
292	return	×
293
NEW 294	message = await self._send_message(	×
295	self.objname if self.objname else self._current_player,
296	PROPERTIES_INTERFACE,
297	MPRIS_PATH,
298	"GetAll",
299	"s",
300	[MPRIS_PLAYER],
301	)
302
303	# If we get an error here it will be because the player object doesn't exist
304	if message.message_type != MessageType.METHOD_RETURN:	×
305	self._current_player = None	×
306	self.update("")	×
307	return	×
308
309	if message.body:	×
310	self._current_player = message.sender	×
311	self.parse_message(self.objname, message.body[0], [])	×
312
313	def _set_background_poll(self, poll=True):	3✔
314	if self._background_poll is not None:	×
315	self._background_poll.cancel()	×
316
317	if poll:	×
318	self._background_poll = self.timeout_add(self.poll_interval, self._check_player)	×
319
320	async def get_player_name(self, player):	3✔
NEW 321	message = await self._send_message(	×
322	player,
323	PROPERTIES_INTERFACE,
324	MPRIS_PATH,
325	"Get",
326	"ss",
327	[MPRIS_OBJECT, "Identity"],
328	)
329
330	if message.message_type != MessageType.METHOD_RETURN:	×
331	logger.warning("Could not retrieve identity of player on %s.", player)	×
332	return ""	×
333
334	return message.body[0].value	×
335
336	def parse_message(	3✔
337	self,
338	_interface_name: str,
339	changed_properties: dict[str, Any],
340	_invalidated_properties: list[str],
341	) -> None:
342	"""
343	http://specifications.freedesktop.org/mpris-spec/latest/Track_List_Interface.html#Mapping:Metadata_Map
344	"""
345	if not self.configured:	3✔
346	return	3✔
347
348	if "Metadata" not in changed_properties and "PlaybackStatus" not in changed_properties:	3✔
349	return	×
350
351	self.displaytext = ""	3✔
352
353	metadata = changed_properties.get("Metadata")	3✔
354	if metadata:	3✔
355	self.track_info = self.get_track_info(metadata.value)	3✔
356
357	playbackstatus = getattr(changed_properties.get("PlaybackStatus"), "value", None)	3✔
358	if playbackstatus:	3✔
359	self.is_playing = playbackstatus == "Playing"	3✔
360	self.status = self.prefixes.get(playbackstatus, "{track}")	3✔
361
362	if not self.track_info:	3✔
363	self.track_info = self.no_metadata_text	3✔
364
365	self.displaytext = self.status.format(track=self.track_info)	3✔
366
367	if self.text != self.displaytext:	3✔
368	self.update(self.displaytext)	3✔
369
370	if self.poll_interval:	3✔
371	self._set_background_poll()	×
372
373	def get_track_info(self, metadata: dict[str, Variant]) -> str:	3✔
374	self.metadata = {}	3✔
375	for key in metadata:	3✔
376	new_key = key	3✔
377	val = getattr(metadata.get(key), "value", None)	3✔
378	if isinstance(val, str):	3✔
379	self.metadata[new_key] = val	3✔
380	elif isinstance(val, list):	3✔
381	self.metadata[new_key] = self.separator.join(	3✔
382	(y for y in val if isinstance(y, str))
383	)
384
385	return self._formatter.format(self.format, **self.metadata).replace("\n", "")	3✔
386
387	def _player_cmd(self, cmd: str) -> None:	3✔
388	if self._current_player is None:	×
389	return	×
390
391	task = asyncio.create_task(self._send_player_cmd(cmd))	×
392	task.add_done_callback(self._task_callback)	×
393
394	async def _send_player_cmd(self, cmd: str) -> Message \| None:	3✔
NEW 395	message = await self._send_message(	×
396	self._current_player,
397	MPRIS_PLAYER,
398	MPRIS_PATH,
399	cmd,
400	"",
401	[],
402	)
403
404	return message	×
405
406	def _task_callback(self, task: asyncio.Task) -> None:	3✔
407	message = task.result()	×
408
409	# This happens if we can't connect to dbus. Logger call is made
410	# elsewhere so we don't need to do any more here.
411	if message is None:	×
412	return	×
413
414	if message.message_type != MessageType.METHOD_RETURN:	×
415	logger.warning("Unable to send command to player.")	×
416
417	@expose_command()	3✔
418	def play_pause(self) -> None:	3✔
419	"""Toggle the playback status."""
420	self._player_cmd("PlayPause")	×
421
422	@expose_command()	3✔
423	def next(self) -> None:	3✔
424	"""Play the next track."""
425	self._player_cmd("Next")	×
426
427	@expose_command()	3✔
428	def previous(self) -> None:	3✔
429	"""Play the previous track."""
430	self._player_cmd("Previous")	×
431
432	@expose_command()	3✔
433	def stop(self) -> None:	3✔
434	"""Stop playback."""
435	self._player_cmd("Stop")	×
436
437	@expose_command()	3✔
438	def info(self):	3✔
439	"""What's the current state of the widget?"""
440	d = base._TextBox.info(self)	3✔
441	d.update(dict(isplaying=self.is_playing, player=self.player))	3✔
442	return d	3✔

qtile / qtile / 3886830687

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