Skip to content

Commit af5b178

Browse files
hehe boi
0 parents  commit af5b178

7 files changed

Lines changed: 610 additions & 0 deletions

File tree

.gitignore

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
# Python-generated files
2+
__pycache__/
3+
*.py[oc]
4+
build/
5+
dist/
6+
wheels/
7+
*.egg-info
8+
9+
# Virtual environments
10+
.venv
11+
12+
*build
13+
*dist
14+
*.bin

.python-version

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
3.13

README.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
# Python Deafen
2+
3+
Mutes/unmutes Discord using PulseAudio.

discord_audio_controller.py

Lines changed: 262 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,262 @@
1+
import subprocess
2+
import re
3+
from typing import List, Dict, Optional
4+
5+
6+
class DiscordAudioController:
7+
"""Controller for managing Discord audio streams through PulseAudio"""
8+
9+
def __init__(self):
10+
self.discord_patterns = [
11+
r"WEBRTC VoiceEngine",
12+
r"Discord",
13+
r"discord",
14+
r"playStream",
15+
r"recStream"
16+
]
17+
18+
def _run_command(self, command: List[str]) -> Optional[str]:
19+
"""Execute a command and return the result"""
20+
try:
21+
result = subprocess.run(command, capture_output=True, text=True, check=True)
22+
return result.stdout
23+
except subprocess.CalledProcessError as e:
24+
print(f"Command failed: {' '.join(command)} - {e}")
25+
return None
26+
27+
def _get_sink_inputs(self) -> List[Dict[str, str]]:
28+
"""Get all PulseAudio sink-inputs (playback streams)"""
29+
output = self._run_command(["pactl", "list", "sink-inputs"])
30+
if not output:
31+
return []
32+
33+
sinks = []
34+
current_sink = {}
35+
36+
for line in output.split('\n'):
37+
line = line.strip()
38+
39+
if line.startswith("Sink Input #"):
40+
if current_sink:
41+
sinks.append(current_sink)
42+
sink_id = re.search(r'#(\d+)', line)
43+
current_sink = {
44+
"id": sink_id.group(1) if sink_id else "",
45+
"type": "sink-input"
46+
}
47+
48+
elif "application.name" in line:
49+
match = re.search(r'application\.name = "([^"]*)"', line)
50+
if match:
51+
current_sink["app_name"] = match.group(1)
52+
53+
elif line.startswith("Mute:"):
54+
current_sink["muted"] = "yes" in line.lower()
55+
56+
elif line.startswith("Volume:"):
57+
volume_match = re.search(r'(\d+)%', line)
58+
if volume_match:
59+
current_sink["volume"] = int(volume_match.group(1))
60+
61+
if current_sink:
62+
sinks.append(current_sink)
63+
64+
return sinks
65+
66+
def _get_source_outputs(self) -> List[Dict[str, str]]:
67+
"""Get all PulseAudio source-outputs (recording streams)"""
68+
output = self._run_command(["pactl", "list", "source-outputs"])
69+
if not output:
70+
return []
71+
72+
sources = []
73+
current_source = {}
74+
75+
for line in output.split('\n'):
76+
line = line.strip()
77+
78+
if line.startswith("Source Output #"):
79+
if current_source:
80+
sources.append(current_source)
81+
source_id = re.search(r'#(\d+)', line)
82+
current_source = {
83+
"id": source_id.group(1) if source_id else "",
84+
"type": "source-output"
85+
}
86+
87+
elif "application.name" in line:
88+
match = re.search(r'application\.name = "([^"]*)"', line)
89+
if match:
90+
current_source["app_name"] = match.group(1)
91+
92+
elif line.startswith("Mute:"):
93+
current_source["muted"] = "yes" in line.lower()
94+
95+
if current_source:
96+
sources.append(current_source)
97+
98+
return sources
99+
100+
def _find_discord_streams(self) -> List[Dict[str, str]]:
101+
"""Find all Discord-related streams (both playback and recording)"""
102+
all_streams = self._get_sink_inputs() + self._get_source_outputs()
103+
discord_streams = []
104+
105+
for stream in all_streams:
106+
app_name = stream.get("app_name", "").lower()
107+
108+
# Check against Discord patterns
109+
for pattern in self.discord_patterns:
110+
if pattern.lower() in app_name:
111+
discord_streams.append(stream)
112+
break
113+
114+
return discord_streams
115+
116+
def get_status(self) -> Dict:
117+
"""Get current status of Discord streams"""
118+
discord_streams = self._find_discord_streams()
119+
120+
if not discord_streams:
121+
return {
122+
"success": True,
123+
"found": False,
124+
"message": "No Discord streams found",
125+
"streams": []
126+
}
127+
128+
streams_info = []
129+
for stream in discord_streams:
130+
stream_info = {
131+
"id": stream.get("id"),
132+
"type": "playback" if stream.get("type") == "sink-input" else "recording",
133+
"app_name": stream.get("app_name", "Unknown"),
134+
"muted": stream.get("muted", False),
135+
"volume": stream.get("volume", None)
136+
}
137+
streams_info.append(stream_info)
138+
139+
return {
140+
"success": True,
141+
"found": True,
142+
"message": f"Found {len(discord_streams)} Discord streams",
143+
"streams": streams_info
144+
}
145+
146+
def toggle_mute(self) -> Dict:
147+
"""Toggle mute state for all Discord streams"""
148+
discord_streams = self._find_discord_streams()
149+
150+
if not discord_streams:
151+
return {
152+
"success": False,
153+
"message": "No Discord streams found",
154+
"streams_affected": 0
155+
}
156+
157+
# Determine target state based on first stream
158+
first_stream = discord_streams[0]
159+
is_currently_muted = first_stream.get("muted", False)
160+
target_state = "0" if is_currently_muted else "1" # 0 = unmute, 1 = mute
161+
action = "unmuted" if is_currently_muted else "muted"
162+
163+
results = []
164+
success_count = 0
165+
166+
for stream in discord_streams:
167+
stream_id = stream.get("id")
168+
stream_type = stream.get("type")
169+
app_name = stream.get("app_name", "Unknown")
170+
171+
if not stream_id or not stream_type:
172+
continue
173+
174+
# Choose correct command based on stream type
175+
if stream_type == "sink-input":
176+
cmd = ["pactl", "set-sink-input-mute", stream_id, target_state]
177+
stream_desc = "playback"
178+
elif stream_type == "source-output":
179+
cmd = ["pactl", "set-source-output-mute", stream_id, target_state]
180+
stream_desc = "recording"
181+
else:
182+
continue
183+
184+
result = self._run_command(cmd)
185+
stream_result = {
186+
"id": stream_id,
187+
"type": stream_desc,
188+
"app_name": app_name,
189+
"success": result is not None,
190+
"action": action
191+
}
192+
193+
if result is not None:
194+
success_count += 1
195+
196+
results.append(stream_result)
197+
198+
return {
199+
"success": success_count > 0,
200+
"message": f"Successfully {action} {success_count}/{len(discord_streams)} streams",
201+
"action": action,
202+
"streams_affected": success_count,
203+
"total_streams": len(discord_streams),
204+
"results": results
205+
}
206+
207+
def set_mute(self, mute: bool) -> Dict:
208+
"""Set mute state for all Discord streams"""
209+
discord_streams = self._find_discord_streams()
210+
211+
if not discord_streams:
212+
return {
213+
"success": False,
214+
"message": "No Discord streams found",
215+
"streams_affected": 0
216+
}
217+
218+
target_state = "1" if mute else "0" # 1 = mute, 0 = unmute
219+
action = "muted" if mute else "unmuted"
220+
221+
results = []
222+
success_count = 0
223+
224+
for stream in discord_streams:
225+
stream_id = stream.get("id")
226+
stream_type = stream.get("type")
227+
app_name = stream.get("app_name", "Unknown")
228+
229+
if not stream_id or not stream_type:
230+
continue
231+
232+
if stream_type == "sink-input":
233+
cmd = ["pactl", "set-sink-input-mute", stream_id, target_state]
234+
stream_desc = "playback"
235+
elif stream_type == "source-output":
236+
cmd = ["pactl", "set-source-output-mute", stream_id, target_state]
237+
stream_desc = "recording"
238+
else:
239+
continue
240+
241+
result = self._run_command(cmd)
242+
stream_result = {
243+
"id": stream_id,
244+
"type": stream_desc,
245+
"app_name": app_name,
246+
"success": result is not None,
247+
"action": action
248+
}
249+
250+
if result is not None:
251+
success_count += 1
252+
253+
results.append(stream_result)
254+
255+
return {
256+
"success": success_count > 0,
257+
"message": f"Successfully {action} {success_count}/{len(discord_streams)} streams",
258+
"action": action,
259+
"streams_affected": success_count,
260+
"total_streams": len(discord_streams),
261+
"results": results
262+
}

main.py

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
#!/usr/bin/env python3
2+
3+
from flask import Flask, jsonify
4+
from flask_cors import CORS
5+
from discord_audio_controller import DiscordAudioController
6+
7+
8+
app = Flask(__name__)
9+
CORS(app)
10+
11+
12+
discord_controller = DiscordAudioController()
13+
14+
15+
@app.route("/", methods=["GET"])
16+
def index():
17+
"""API information endpoint"""
18+
return jsonify({
19+
"service": "Discord Audio Controller",
20+
"version": "1.0.0",
21+
"endpoints": {
22+
"/status": "GET - Get current Discord audio status",
23+
"/toggle": "POST - Toggle mute state for Discord streams",
24+
"/mute": "POST - Mute all Discord streams",
25+
"/unmute": "POST - Unmute all Discord streams"
26+
}
27+
})
28+
29+
30+
@app.route("/status", methods=["GET"])
31+
def get_status():
32+
"""Get current status of Discord audio streams"""
33+
try:
34+
result = discord_controller.get_status()
35+
return jsonify(result)
36+
except Exception as e:
37+
return jsonify({
38+
"success": False,
39+
"message": f"Error getting status: {str(e)}"
40+
}), 500
41+
42+
43+
@app.route("/toggle", methods=["POST"])
44+
def toggle_mute():
45+
"""Toggle mute state for Discord streams"""
46+
try:
47+
result = discord_controller.toggle_mute()
48+
status_code = 200 if result["success"] else 404
49+
return jsonify(result), status_code
50+
except Exception as e:
51+
return jsonify({
52+
"success": False,
53+
"message": f"Error toggling mute: {str(e)}"
54+
}), 500
55+
56+
57+
@app.route("/mute", methods=["POST"])
58+
def mute_discord():
59+
"""Mute all Discord streams"""
60+
try:
61+
result = discord_controller.set_mute(True)
62+
status_code = 200 if result["success"] else 404
63+
return jsonify(result), status_code
64+
except Exception as e:
65+
return jsonify({
66+
"success": False,
67+
"message": f"Error muting Discord: {str(e)}"
68+
}), 500
69+
70+
71+
@app.route("/unmute", methods=["POST"])
72+
def unmute_discord():
73+
"""Unmute all Discord streams"""
74+
try:
75+
result = discord_controller.set_mute(False)
76+
status_code = 200 if result["success"] else 404
77+
return jsonify(result), status_code
78+
except Exception as e:
79+
return jsonify({
80+
"success": False,
81+
"message": f"Error unmuting Discord: {str(e)}"
82+
}), 500
83+
84+
85+
def create_app():
86+
"""Application factory for the Discord Audio Controller"""
87+
return app
88+
89+
90+
if __name__ == "__main__":
91+
print("Starting Discord Audio Controller Web Server...")
92+
print("Available endpoints:")
93+
print(" GET / - API information")
94+
print(" GET /status - Get Discord audio status")
95+
print(" POST /toggle - Toggle Discord mute state")
96+
print(" POST /mute - Mute Discord")
97+
print(" POST /unmute - Unmute Discord")
98+
print()
99+
100+
# Run the Flask development server
101+
app.run(host="0.0.0.0", port=18498, debug=True)

0 commit comments

Comments
 (0)