Skip to content

Added support to support PipeWire #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
carlocastoldi wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Added support to support PipeWire #1

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 16 additions & 11 deletions blockify/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,11 @@ def start_spotify_if_necessary(self):
def is_localized_pulseaudio(self):
"""Pulseaudio versions below 7.0 are localized."""
localized = False
try:
subprocess.check_call(["pulseaudio"])
except OSError:
log.info("Assuming system is using PipeWire")
return localized
try:
pulseaudio_version_string = subprocess.check_output("pulseaudio --version | awk '{print $2}'", shell=True)
pulseaudio_version = int(pulseaudio_version_string[0])
Expand Down Expand Up @@ -126,7 +131,7 @@ def initialize_mute_method(self):
"""Determine if we can use sinks or have to use alsa."""
try:
devnull = open(os.devnull)
subprocess.check_output(["pacmd", "list-sink-inputs"], stderr=devnull)
subprocess.check_output(["pactl", "list", "sink-inputs"], stderr=devnull)
self.mutemethod = self.pulsesink_mute
log.debug("Mute method is pulse sink.")
except (OSError, subprocess.CalledProcessError):
Expand Down Expand Up @@ -425,14 +430,14 @@ def update_audio_channel_state(self, command, state):
except subprocess.CalledProcessError:
pass

def extract_pulse_sink_status(self, pacmd_out):
def extract_pulse_sink_status(self, pactl_out):
sink_status = ("", "", "") # index, playback_status, muted_value
# Match muted_value and application.process.id values.
pattern = re.compile(r"(?: index|state|muted|application\.process\.id).*?(\w+)")
pattern = re.compile(r"(?: Sink Input #|Corked|Mute|application\.process\.id).*?(\w+)")
# Put valid spotify PIDs in a list
output = pacmd_out.decode("utf-8")
output = pactl_out.decode("utf-8")

spotify_sink_list = [" index: " + i for i in output.split("index: ") if "spotify" in i]
spotify_sink_list = [" Sink Input #" + i for i in output.split("Sink Input #") if "spotify" in i]

if len(spotify_sink_list) and self.spotify_pids:
sink_infos = [pattern.findall(sink) for sink in spotify_sink_list]
Expand All @@ -449,27 +454,27 @@ def extract_pulse_sink_status(self, pacmd_out):
def pulsesink_mute(self, mode):
"""Finds spotify's audio sink and toggles its mute state."""
try:
pacmd_out = subprocess.check_output(["pacmd", "list-sink-inputs"])
pactl_out = subprocess.check_output(["pactl", "list", "sink-inputs"])
except subprocess.CalledProcessError:
log.error("Spotify sink not found. Is Pulse running? Resorting to pulse amixer as mute method.")
self.mutemethod = self.pulse_mute # Fall back to amixer mute.
self.use_interlude_music = False
return

index, playback_state, muted_value = self.extract_pulse_sink_status(pacmd_out)
self.song_status = "Playing" if playback_state == "RUNNING" else "Paused"
index, playback_state, muted_value = self.extract_pulse_sink_status(pactl_out)
self.song_status = "Playing" if playback_state == self.pulse_unmuted_value else "Paused"
self.is_sink_muted = False if muted_value == self.pulse_unmuted_value else True

if index:
if self.is_sink_muted and (mode == 2 or not self.current_song):
log.info("Forcing unmute.")
subprocess.call(["pacmd", "set-sink-input-mute", index, "0"])
subprocess.call(["pactl", "set-sink-input-mute", index, "no"])
elif not self.is_sink_muted and mode == 1:
log.info("Muting {}.".format(self.current_song))
subprocess.call(["pacmd", "set-sink-input-mute", index, "1"])
subprocess.call(["pactl", "set-sink-input-mute", index, "yes"])
elif self.is_sink_muted and not mode:
log.info("Unmuting.")
subprocess.call(["pacmd", "set-sink-input-mute", index, "0"])
subprocess.call(["pactl", "set-sink-input-mute", index, "no"])

def prev(self):
self.dbus.prev()
Expand Down