670 lines
36 KiB
Python
670 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
"""VoicePult: local PipeWire/PulseAudio patch panel, Python standard library only."""
|
||
import argparse
|
||
import copy
|
||
import fcntl
|
||
import hashlib
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import secrets
|
||
import signal
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from urllib.parse import urlparse, parse_qs
|
||
from audio_meter import MeterHub
|
||
from lan_access import LanAccess
|
||
import uuid
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
PREFIX = 'voicepult_'
|
||
MIX = PREFIX + 'mix'
|
||
MIC = PREFIX + 'microphone'
|
||
|
||
|
||
class AudioError(Exception):
|
||
pass
|
||
|
||
|
||
class Pulse:
|
||
def run(self, *args):
|
||
try:
|
||
p = subprocess.run(['pactl', *map(str, args)], capture_output=True,
|
||
text=True, timeout=8, env={**os.environ, 'LC_ALL': 'C.UTF-8'})
|
||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||
raise AudioError('Audioserver nicht erreichbar. PipeWire/PulseAudio und pactl prüfen.') from exc
|
||
if p.returncode:
|
||
raise AudioError(p.stderr.strip() or 'Audiobefehl fehlgeschlagen.')
|
||
return p.stdout.strip()
|
||
|
||
def items(self, kind):
|
||
if kind == 'modules':
|
||
# Some pactl versions omit module indices in their JSON output.
|
||
modules = []
|
||
for line in self.run('list', 'short', 'modules').splitlines():
|
||
fields = line.split('\t')
|
||
if len(fields) >= 3 and fields[0].isdigit():
|
||
modules.append({'index': int(fields[0]), 'name': fields[1], 'argument': fields[2]})
|
||
return modules
|
||
return json.loads(self.run('-f', 'json', 'list', kind))
|
||
|
||
|
||
def app_key(stream):
|
||
props = stream.get('properties', {})
|
||
return props.get('application.process.binary', '') + '|' + props.get('application.name', 'Unbekannte Anwendung')
|
||
|
||
|
||
def is_internal(stream):
|
||
props = stream.get('properties', {})
|
||
return props.get('application.id') == 'de.voicepult' or props.get('media.name', '').startswith('VoicePult')
|
||
|
||
|
||
class Panel:
|
||
def __init__(self, path, pulse=None):
|
||
self.path, self.pulse = Path(path), pulse or Pulse()
|
||
self.lock = threading.RLock()
|
||
self.modules = {} # logical key -> (module id, exact argument string)
|
||
self.originals = {}
|
||
self.route_states = {}
|
||
self.error = None
|
||
self.config = {'targets': [], 'channels': [], 'assignments': {}, 'output': '', 'input': '', 'mic_enabled': False, 'mix_volume': 100, 'mix_muted': False, 'theme': 'light', 'wave_gain': 8, 'mic_channels': [], 'scope_source': MIC, 'scope_enabled': True}
|
||
if self.path.exists():
|
||
self.config.update(json.loads(self.path.read_text()))
|
||
|
||
self.config['assignments'] = {k: ([v] if isinstance(v, str) else v)
|
||
for k, v in self.config['assignments'].items() if v}
|
||
|
||
@staticmethod
|
||
def bus(app):
|
||
return PREFIX + 'app_' + hashlib.sha256(app.encode()).hexdigest()[:16]
|
||
|
||
def save(self):
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
temp = self.path.with_suffix('.tmp')
|
||
temp.write_text(json.dumps(self.config, ensure_ascii=False, indent=2))
|
||
temp.replace(self.path)
|
||
|
||
def ensure_module(self, key, kind, *args):
|
||
if key not in self.modules:
|
||
mid = int(self.pulse.run('load-module', kind, *args))
|
||
self.modules[key] = (mid, ' '.join(args))
|
||
|
||
def unload(self, key):
|
||
record = self.modules.get(key)
|
||
if record:
|
||
if key.startswith('route:'):
|
||
self.set_route_enabled(key, False)
|
||
self.route_states.pop(key, None)
|
||
self.pulse.run('unload-module', record[0])
|
||
del self.modules[key]
|
||
|
||
def fade(self, kind, index, start, end):
|
||
# Keep transitions short, but avoid a full-level discontinuity.
|
||
for step in range(1, 9):
|
||
value = round(start + (end - start) * step / 8)
|
||
self.pulse.run('set-' + kind + '-volume', index, f'{value}%')
|
||
if step < 8:
|
||
time.sleep(0.008)
|
||
|
||
def sink_level(self, sink, volume, muted, previous):
|
||
if muted != previous['muted']:
|
||
if muted:
|
||
self.fade('sink', sink, previous['volume'], 0)
|
||
self.pulse.run('set-sink-mute', sink, 1)
|
||
self.pulse.run('set-sink-volume', sink, f'{volume}%')
|
||
else:
|
||
self.pulse.run('set-sink-volume', sink, '0%')
|
||
self.pulse.run('set-sink-mute', sink, 0)
|
||
self.fade('sink', sink, 0, volume)
|
||
else:
|
||
self.pulse.run('set-sink-volume', sink, f'{volume}%')
|
||
self.pulse.run('set-sink-mute', sink, int(muted))
|
||
|
||
def set_route_enabled(self, key, enabled):
|
||
record = self.modules.get(key)
|
||
if not record or self.route_states.get(key) == (record[0], enabled):
|
||
return
|
||
stream = None
|
||
for attempt in range(20):
|
||
stream = next((s for s in self.pulse.items('sink-inputs')
|
||
if str(s.get('owner_module')) == str(record[0]) and is_internal(s)), None)
|
||
if stream:
|
||
break
|
||
time.sleep(0.01)
|
||
if not stream:
|
||
raise AudioError('Audioverbindung ist noch nicht bereit. Bitte erneut versuchen.')
|
||
idx = stream['index']
|
||
if enabled:
|
||
self.pulse.run('set-sink-input-volume', idx, '0%')
|
||
self.pulse.run('set-sink-input-mute', idx, 0)
|
||
self.fade('sink-input', idx, 0, 100)
|
||
else:
|
||
if not stream.get('mute', False):
|
||
values = [v['value'] * 100 / 65536 for v in stream.get('volume', {}).values()]
|
||
self.fade('sink-input', idx, max(values, default=100), 0)
|
||
self.pulse.run('set-sink-input-mute', idx, 1)
|
||
self.route_states[key] = (record[0], enabled)
|
||
|
||
def loop(self, key, source, sink, enabled=True):
|
||
# A disabled, previously used route stays connected and silent. This avoids
|
||
# graph rebuilds and buffer transients each time a destination is toggled.
|
||
if not enabled and key not in self.modules:
|
||
return
|
||
new = key not in self.modules
|
||
self.ensure_module(key, 'module-loopback', f'source={source}', f'sink={sink}',
|
||
'latency_msec=30', 'source_dont_move=true', 'sink_dont_move=true',
|
||
'sink_input_properties="application.id=de.voicepult state.restore-props=false node.param.Props=\'{ mute = true channelVolumes = [ 0.0 0.0 ] }\'"',
|
||
'source_output_properties=application.id=de.voicepult')
|
||
if new:
|
||
self.set_route_enabled(key, False)
|
||
time.sleep(0.04) # Let the initially silent loopback fill its buffer.
|
||
self.set_route_enabled(key, enabled)
|
||
|
||
def recover(self):
|
||
"""Remove only our modules left behind by an interrupted prior process."""
|
||
owned = [m for m in self.pulse.items('modules') if
|
||
(m['name'] == 'module-null-sink' and m['argument'].startswith('sink_name=voicepult_')) or
|
||
(m['name'] == 'module-remap-source' and (m['argument'].startswith('master=voicepult_mix.monitor ') or m['argument'].startswith('master=voicepult_target_'))) or
|
||
(m['name'] == 'module-loopback' and 'application.id=de.voicepult' in m['argument'])]
|
||
if not owned:
|
||
return
|
||
sinks = self.pulse.items('sinks')
|
||
physical = {s['name'] for s in sinks if not s['name'].startswith(PREFIX)}
|
||
virtual = {s['index'] for s in sinks if s['name'].startswith(PREFIX)}
|
||
for stream in self.pulse.items('sink-inputs'):
|
||
if stream['sink'] in virtual and not is_internal(stream):
|
||
self.restore(stream, physical)
|
||
for module in reversed(owned):
|
||
self.pulse.run('unload-module', module['index'])
|
||
|
||
def reconcile(self):
|
||
# A restarted audio server invalidates ids. Never unload a recycled module id.
|
||
live = {int(m['index']): m for m in self.pulse.items('modules') if 'index' in m}
|
||
self.modules = {k: v for k, v in self.modules.items()
|
||
if v[0] in live and live[v[0]].get('argument', '') == v[1]}
|
||
self.route_states = {k: v for k, v in self.route_states.items() if k in self.modules and self.modules[k][0] == v[0]}
|
||
sinks = self.pulse.items('sinks')
|
||
sources = self.pulse.items('sources')
|
||
outputs = [s for s in sinks if not s['name'].startswith(PREFIX)]
|
||
inputs = [s for s in sources if not s['name'].startswith(PREFIX)
|
||
and s.get('properties', {}).get('device.class') != 'monitor'
|
||
and not s['name'].endswith('.monitor')]
|
||
if not self.config['output'] and outputs:
|
||
default = self.pulse.run('get-default-sink')
|
||
self.config['output'] = next((s['name'] for s in outputs if s['name'] == default), outputs[0]['name'])
|
||
if not self.config['input'] and inputs:
|
||
default = self.pulse.run('get-default-source')
|
||
self.config['input'] = next((s['name'] for s in inputs if s['name'] == default), inputs[0]['name'])
|
||
new_mix = 'mix' not in self.modules
|
||
self.ensure_module('mix', 'module-null-sink', f'sink_name={MIX}',
|
||
'sink_properties=device.description=VoicePult_Mix', 'rate=48000', 'channels=2')
|
||
if new_mix:
|
||
self.pulse.run('set-sink-volume', MIX, f'{self.config["mix_volume"]}%')
|
||
self.pulse.run('set-sink-mute', MIX, int(self.config['mix_muted']))
|
||
self.ensure_module('mic', 'module-remap-source', f'master={MIX}.monitor', f'source_name={MIC}',
|
||
'source_properties="device.description=\'VoicePult Mikrofon\' device.class=sound device.icon_name=audio-input-microphone"')
|
||
for target in self.config['targets']:
|
||
tid = target['id']
|
||
sink = PREFIX + 'target_' + tid
|
||
new = 'target:' + tid not in self.modules
|
||
self.ensure_module('target:' + tid, 'module-null-sink', f'sink_name={sink}',
|
||
f'sink_properties=device.description=VoicePult_Target_{tid}', 'rate=48000', 'channels=2')
|
||
if new:
|
||
self.pulse.run('set-sink-volume', sink, f'{target["volume"]}%')
|
||
self.pulse.run('set-sink-mute', sink, int(target['muted']))
|
||
self.ensure_module('target-source:' + tid, 'module-remap-source', f'master={sink}.monitor',
|
||
f'source_name={sink}_mic',
|
||
'source_properties=' + json.dumps('device.description=' + json.dumps('VoicePult ' + target['name'], ensure_ascii=False) + ' device.class=sound device.icon_name=audio-input-microphone', ensure_ascii=False))
|
||
desired = set()
|
||
for ch in self.config['channels']:
|
||
name = PREFIX + ch['id']
|
||
key = 'channel:' + ch['id']
|
||
new = key not in self.modules
|
||
self.ensure_module(key, 'module-null-sink', f'sink_name={name}',
|
||
f'sink_properties=device.description=VoicePult_{ch["id"]}', 'rate=48000', 'channels=2')
|
||
if new:
|
||
self.pulse.run('set-sink-volume', name, f'{ch["volume"]}%')
|
||
self.pulse.run('set-sink-mute', name, int(ch['muted']))
|
||
if any(s['name'] == self.config['output'] for s in outputs):
|
||
route = 'route:' + ch['id'] + ':' + self.config['output']
|
||
desired.add(route)
|
||
self.loop(route, name + '.monitor', self.config['output'], ch['monitor'])
|
||
for target in self.config['targets']:
|
||
tid = target['id']
|
||
route = 'route:' + ch['id'] + ':target:' + tid
|
||
desired.add(route)
|
||
self.loop(route, name + '.monitor', PREFIX + 'target_' + tid, tid in ch.get('targets', []))
|
||
route = 'route:' + ch['id'] + ':mic'
|
||
desired.add(route)
|
||
self.loop(route, name + '.monitor', MIX, ch['microphone'])
|
||
if any(s['name'] == self.config['input'] for s in inputs):
|
||
route = 'route:input:' + self.config['input']
|
||
desired.add(route)
|
||
self.loop(route, self.config['input'], MIX, self.config['mic_enabled'])
|
||
for channel in self.config['channels']:
|
||
cid = channel['id']
|
||
route = 'route:physical:' + cid + ':' + self.config['input']
|
||
desired.add(route)
|
||
self.loop(route, self.config['input'], PREFIX + cid, cid in self.config['mic_channels'])
|
||
buses = set()
|
||
for app, channels in self.config['assignments'].items():
|
||
if len(channels) > 1:
|
||
bus = self.bus(app)
|
||
buses.add('bus:' + bus)
|
||
self.ensure_module('bus:' + bus, 'module-null-sink', f'sink_name={bus}',
|
||
'sink_properties=device.description=VoicePult_App', 'rate=48000', 'channels=2')
|
||
for cid in channels:
|
||
route = 'route:app:' + bus + ':to:' + cid
|
||
desired.add(route)
|
||
self.loop(route, bus + '.monitor', PREFIX + cid)
|
||
for key in list(self.modules):
|
||
if key.startswith('route:') and key not in desired:
|
||
self.unload(key)
|
||
sink_names = {s['index']: s['name'] for s in self.pulse.items('sinks')}
|
||
streams = self.pulse.items('sink-inputs')
|
||
# Stream indices may be reused after disconnect; discard stale restore entries.
|
||
identities = {s['index']: self.identity(s) for s in streams}
|
||
self.originals = {idx: rec for idx, rec in self.originals.items() if identities.get(idx) == rec[0]}
|
||
for stream in streams:
|
||
if is_internal(stream):
|
||
continue
|
||
target = self.config['assignments'].get(app_key(stream))
|
||
if target:
|
||
name = PREFIX + target[0] if len(target) == 1 else self.bus(app_key(stream))
|
||
current = sink_names.get(stream['sink'], '')
|
||
if current != name:
|
||
if stream['index'] not in self.originals and not current.startswith(PREFIX):
|
||
self.originals[stream['index']] = (self.identity(stream), current)
|
||
self.pulse.run('move-sink-input', stream['index'], name)
|
||
for key in list(self.modules):
|
||
if key.startswith('bus:') and key not in buses:
|
||
self.unload(key)
|
||
self.error = None
|
||
|
||
@staticmethod
|
||
def identity(stream):
|
||
p = stream.get('properties', {})
|
||
return (app_key(stream), p.get('object.serial'), p.get('application.process.id'))
|
||
|
||
def restore(self, stream, sinks):
|
||
old = self.originals.get(stream['index'])
|
||
dest = old[1] if old and old[0] == self.identity(stream) else self.config['output']
|
||
if dest not in sinks:
|
||
dest = next(iter(sinks), None)
|
||
if not dest:
|
||
raise AudioError('Kein physischer Ausgang zum Zurücksetzen verfügbar.')
|
||
self.pulse.run('move-sink-input', stream['index'], dest)
|
||
self.originals.pop(stream['index'], None)
|
||
|
||
def meter_sources(self):
|
||
return ({MIC} | {PREFIX + c['id'] + '.monitor' for c in self.config['channels']}
|
||
| {PREFIX + 'target_' + t['id'] + '_mic' for t in self.config['targets']})
|
||
|
||
def saved_apps(self, live=None):
|
||
live = live or {}
|
||
result = []
|
||
for key, channels in self.config['assignments'].items():
|
||
if not channels:
|
||
continue
|
||
app = {'id': key, 'name': key.partition('|')[2] or key.partition('|')[0],
|
||
'active': False, 'streams': 0, 'destinations': []}
|
||
app.update(live.get(key, {}))
|
||
app.update(channels=list(channels), present=key in live)
|
||
result.append(app)
|
||
return sorted(result, key=lambda app: (app['name'].lower(), app['id']))
|
||
|
||
def state(self):
|
||
with self.lock:
|
||
try:
|
||
sinks = self.pulse.items('sinks')
|
||
sources = self.pulse.items('sources')
|
||
streams = self.pulse.items('sink-inputs')
|
||
devices = lambda xs: [{'id': x['name'], 'name': x.get('description', x['name'])} for x in xs]
|
||
apps = {}
|
||
sink_names = {s['index']: s['name'] for s in sinks}
|
||
for s in streams:
|
||
if is_internal(s):
|
||
continue
|
||
key = app_key(s)
|
||
app = apps.setdefault(key, {'id': key, 'name': s.get('properties', {}).get('application.name', 'Anwendung'),
|
||
'active': False, 'streams': 0, 'channels': self.config['assignments'].get(key, []), 'channel': next(iter(self.config['assignments'].get(key, [])), ''), 'destinations': []})
|
||
app['active'] |= not s.get('corked', False)
|
||
app['streams'] += 1
|
||
dest = sink_names.get(s['sink'], '')
|
||
if dest not in app['destinations']:
|
||
app['destinations'].append(dest)
|
||
return {**copy.deepcopy(self.config), 'online': not bool(self.error), 'error': self.error,
|
||
'outputs': devices([s for s in sinks if not s['name'].startswith(PREFIX)]),
|
||
'inputs': devices([s for s in sources if not s['name'].startswith(PREFIX) and
|
||
s.get('properties', {}).get('device.class') != 'monitor' and not s['name'].endswith('.monitor')]),
|
||
'saved_apps': self.saved_apps(apps),
|
||
'apps': sorted(apps.values(), key=lambda a: a['name'].lower()), 'virtual_mic': 'VoicePult Mikrofon',
|
||
'mic_available': any(s['name'] == MIC for s in sources),
|
||
'mic_consumers': sorted({s.get('properties', {}).get('application.name', 'Anwendung')
|
||
for s in self.pulse.items('source-outputs')
|
||
if not is_internal(s) and s['source'] in
|
||
{src['index'] for src in sources if src['name'] == MIC}})}
|
||
except (AudioError, ValueError) as exc:
|
||
return {**copy.deepcopy(self.config), 'online': False, 'error': str(exc),
|
||
'outputs': [], 'inputs': [], 'apps': [], 'saved_apps': self.saved_apps(), 'virtual_mic': 'VoicePult Mikrofon',
|
||
'mic_available': False, 'mic_consumers': []}
|
||
|
||
def action(self, data):
|
||
with self.lock:
|
||
previous = copy.deepcopy(self.config)
|
||
try:
|
||
return self._action(data)
|
||
except (ValueError, TypeError):
|
||
self.config = previous
|
||
raise
|
||
|
||
def _action(self, data):
|
||
with self.lock:
|
||
action = data.get('action')
|
||
if action in ('toggle_channel', 'toggle_target'):
|
||
collection = 'channels' if action == 'toggle_channel' else 'targets'
|
||
item = next((c for c in self.config[collection] if c['id'] == data.get('id')), None)
|
||
if item is None:
|
||
raise ValueError('Audiokanal nicht gefunden.')
|
||
return self._action({'action': 'channel' if collection == 'channels' else 'target',
|
||
'id': item['id'], 'muted': not item['muted']})
|
||
if action == 'target_create':
|
||
name = str(data.get('name', '')).strip()
|
||
if not 1 <= len(name) <= 40 or any(ord(c) < 32 for c in name):
|
||
raise ValueError('Mikrofonname muss 1–40 druckbare Zeichen enthalten.')
|
||
if len(self.config['targets']) >= 16:
|
||
raise ValueError('Maximal 16 zusätzliche Mikrofone möglich.')
|
||
if any(t['name'].casefold() == name.casefold() for t in self.config['targets']):
|
||
raise ValueError('Dieser Mikrofonname ist bereits vergeben.')
|
||
self.config['targets'].append({'id': uuid.uuid4().hex[:12], 'name': name, 'volume': 100, 'muted': False})
|
||
elif action in ('target', 'target_delete'):
|
||
target = next((t for t in self.config['targets'] if t['id'] == data.get('id')), None)
|
||
if target is None:
|
||
raise ValueError('Mikrofonziel nicht gefunden.')
|
||
tid = target['id']
|
||
if action == 'target_delete':
|
||
for ch in self.config['channels']:
|
||
if tid in ch.get('targets', []):
|
||
ch['targets'] = [i for i in ch['targets'] if i != tid]
|
||
for key in list(self.modules):
|
||
if key.startswith('route:') and key.endswith(':target:' + tid):
|
||
self.unload(key)
|
||
self.unload('target-source:' + tid)
|
||
self.unload('target:' + tid)
|
||
self.config['targets'].remove(target)
|
||
if self.config['scope_source'] == PREFIX + 'target_' + tid + '_mic':
|
||
self.config['scope_source'] = MIC
|
||
else:
|
||
volume = data.get('volume', target['volume'])
|
||
muted = data.get('muted', target['muted'])
|
||
if type(volume) is not int or not 0 <= volume <= 100 or type(muted) is not bool:
|
||
raise ValueError('Ungültige Mikrofon-Lautstärke oder Stummschaltung.')
|
||
self.sink_level(PREFIX + 'target_' + tid, volume, muted, target)
|
||
target.update(volume=volume, muted=muted)
|
||
elif action == 'create':
|
||
name = str(data.get('name', '')).strip()
|
||
if not name or len(name) > 40:
|
||
raise ValueError('Kanalname muss 1–40 Zeichen enthalten.')
|
||
if len(self.config['channels']) >= 16:
|
||
raise ValueError('Maximal 16 Kanäle möglich.')
|
||
ch = {'id': uuid.uuid4().hex[:12], 'name': name, 'volume': 100, 'muted': False,
|
||
'monitor': True, 'microphone': False}
|
||
self.config['channels'].append(ch)
|
||
elif action in ('channel', 'delete'):
|
||
ch = next((c for c in self.config['channels'] if c['id'] == data.get('id')), None)
|
||
if not ch:
|
||
raise ValueError('Kanal nicht gefunden.')
|
||
if action == 'delete':
|
||
sinks = {s['name'] for s in self.pulse.items('sinks') if not s['name'].startswith(PREFIX)}
|
||
channel_sink = next((s['index'] for s in self.pulse.items('sinks') if s['name'] == PREFIX + ch['id']), None)
|
||
for s in self.pulse.items('sink-inputs'):
|
||
if s['sink'] == channel_sink and not is_internal(s):
|
||
self.restore(s, sinks)
|
||
updated = {k: [cid for cid in v if cid != ch['id']] for k, v in self.config['assignments'].items()}
|
||
for stream in self.pulse.items('sink-inputs'):
|
||
key = app_key(stream)
|
||
if key in updated and not updated[key] and not is_internal(stream) and stream['sink'] != channel_sink:
|
||
self.restore(stream, sinks)
|
||
self.config['assignments'] = {k: v for k, v in updated.items() if v}
|
||
self.config['mic_channels'] = [cid for cid in self.config['mic_channels'] if cid != ch['id']]
|
||
if self.config['scope_source'] == PREFIX + ch['id'] + '.monitor':
|
||
self.config['scope_source'] = MIC
|
||
for key in list(self.modules):
|
||
if key.startswith('route:' + ch['id'] + ':') or key.endswith(':to:' + ch['id']) or key.startswith('route:physical:' + ch['id'] + ':'):
|
||
self.unload(key)
|
||
self.unload('channel:' + ch['id'])
|
||
self.config['channels'].remove(ch)
|
||
else:
|
||
old_channel = copy.deepcopy(ch)
|
||
if 'targets' in data:
|
||
targets = data['targets']
|
||
if not isinstance(targets, list) or len(targets) > 16 or any(
|
||
not isinstance(t, str) or t not in [x['id'] for x in self.config['targets']] for t in targets):
|
||
raise ValueError('Ungültige Mikrofonziele.')
|
||
ch['targets'] = list(dict.fromkeys(targets))
|
||
for flag in ('monitor', 'microphone', 'muted'):
|
||
if flag in data:
|
||
if type(data[flag]) is not bool:
|
||
raise ValueError('Ungültiger Schalterwert.')
|
||
ch[flag] = data[flag]
|
||
if 'volume' in data:
|
||
if type(data['volume']) is not int or not 0 <= data['volume'] <= 100:
|
||
raise ValueError('Lautstärke muss zwischen 0 und 100 liegen.')
|
||
ch['volume'] = data['volume']
|
||
if 'name' in data:
|
||
name = str(data['name']).strip()
|
||
if not 1 <= len(name) <= 40:
|
||
raise ValueError('Kanalname muss 1–40 Zeichen enthalten.')
|
||
ch['name'] = name
|
||
self.sink_level(PREFIX + ch['id'], ch['volume'], ch['muted'], old_channel)
|
||
elif action in ('assign', 'mic_channels'):
|
||
channels = data.get('channels', [data['channel']] if data.get('channel') else [])
|
||
if not isinstance(channels, list) or len(channels) > 16 or any(
|
||
not isinstance(cid, str) or cid not in [c['id'] for c in self.config['channels']] for cid in channels):
|
||
raise ValueError('Ungültige Kanalauswahl.')
|
||
channels = list(dict.fromkeys(channels))
|
||
if action == 'mic_channels':
|
||
self.config['mic_channels'] = channels
|
||
else:
|
||
app = data.get('app')
|
||
if not isinstance(app, str) or not app or len(app) > 512:
|
||
raise ValueError('Ungültige Anwendung.')
|
||
if channels:
|
||
self.config['assignments'][app] = channels
|
||
else:
|
||
sinks = {s['name'] for s in self.pulse.items('sinks') if not s['name'].startswith(PREFIX)}
|
||
for stream in self.pulse.items('sink-inputs'):
|
||
if app_key(stream) == app:
|
||
self.restore(stream, sinks)
|
||
self.config['assignments'].pop(app, None)
|
||
elif action == 'preferences':
|
||
theme = data.get('theme', self.config['theme'])
|
||
gain = data.get('wave_gain', self.config['wave_gain'])
|
||
if theme not in ('light', 'dark') or type(gain) is not int or not 1 <= gain <= 32:
|
||
raise ValueError('Ungültige Anzeigeeinstellung.')
|
||
source = data.get('scope_source', self.config['scope_source'])
|
||
enabled = data.get('scope_enabled', self.config['scope_enabled'])
|
||
allowed = self.meter_sources()
|
||
if not isinstance(source, str) or source not in allowed or type(enabled) is not bool:
|
||
raise ValueError('Ungültige Wellenform-Einstellung.')
|
||
self.config.update(theme=theme, wave_gain=gain, scope_source=source, scope_enabled=enabled)
|
||
elif action == 'mix':
|
||
volume = data.get('volume', self.config['mix_volume'])
|
||
muted = data.get('muted', self.config['mix_muted'])
|
||
if type(volume) is not int or not 0 <= volume <= 100 or type(muted) is not bool:
|
||
raise ValueError('Ungültige Mix-Lautstärke oder Stummschaltung.')
|
||
self.sink_level(MIX, volume, muted, {'volume': self.config['mix_volume'], 'muted': self.config['mix_muted']})
|
||
self.config.update(mix_volume=volume, mix_muted=muted)
|
||
elif action == 'settings':
|
||
state = self.state()
|
||
for field, collection in [('output', 'outputs'), ('input', 'inputs')]:
|
||
if field in data:
|
||
if data[field] not in [d['id'] for d in state[collection]]:
|
||
raise ValueError('Audiogerät nicht verfügbar.')
|
||
self.config[field] = data[field]
|
||
if 'mic_enabled' in data:
|
||
if type(data['mic_enabled']) is not bool:
|
||
raise ValueError('Ungültiger Schalterwert.')
|
||
self.config['mic_enabled'] = data['mic_enabled']
|
||
else:
|
||
raise ValueError('Unbekannte Aktion.')
|
||
self.save()
|
||
self.reconcile()
|
||
return self.state()
|
||
|
||
def close(self):
|
||
with self.lock:
|
||
# Validate ownership against live modules before unloading.
|
||
live = {int(m['index']): m for m in self.pulse.items('modules') if 'index' in m}
|
||
self.modules = {k: v for k, v in self.modules.items() if v[0] in live and live[v[0]].get('argument', '') == v[1]}
|
||
sinks = self.pulse.items('sinks')
|
||
physical = {s['name'] for s in sinks if not s['name'].startswith(PREFIX)}
|
||
virtual = {s['index'] for s in sinks if s['name'].startswith(PREFIX)}
|
||
for s in self.pulse.items('sink-inputs'):
|
||
if s['sink'] in virtual and not is_internal(s):
|
||
self.restore(s, physical)
|
||
for key in reversed(list(self.modules)):
|
||
try:
|
||
self.unload(key)
|
||
except AudioError:
|
||
pass
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('--port', type=int, default=8765)
|
||
parser.add_argument('--config', default=str(ROOT / '.voicepult' / 'config.json'))
|
||
parser.add_argument('--lan', action='store_true', help='Zugriff im lokalen Netzwerk mit Zugangscode aktivieren')
|
||
args = parser.parse_args()
|
||
# One instance per user, even with different config paths.
|
||
runtime = Path(os.environ.get('XDG_RUNTIME_DIR', '/tmp'))
|
||
lockfile = (runtime / f'voicepult-{os.getuid()}.lock').open('w')
|
||
try:
|
||
fcntl.flock(lockfile, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except BlockingIOError:
|
||
parser.exit(1, 'VoicePult läuft bereits. Öffne http://127.0.0.1:8765\n')
|
||
panel = Panel(args.config)
|
||
access = LanAccess(args.config, args.lan, args.port)
|
||
meters = MeterHub()
|
||
token = secrets.token_urlsafe(32)
|
||
stop = threading.Event()
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
def log_message(self, fmt, *values):
|
||
if values and str(values[1] if len(values) > 1 else '') != '200':
|
||
super().log_message(fmt, *values)
|
||
|
||
def send(self, code, payload, mime='application/json; charset=utf-8'):
|
||
if not isinstance(payload, bytes):
|
||
payload = json.dumps(payload, ensure_ascii=False).encode()
|
||
self.send_response(code)
|
||
self.send_header('Content-Type', mime)
|
||
self.send_header('Content-Length', str(len(payload)))
|
||
self.send_header('Cache-Control', 'no-store')
|
||
self.send_header('X-Content-Type-Options', 'nosniff')
|
||
self.send_header('Content-Security-Policy', "default-src 'self'; style-src 'self'; script-src 'self'; img-src 'self' data:; frame-ancestors 'none'")
|
||
self.end_headers()
|
||
try:
|
||
self.wfile.write(payload)
|
||
except (BrokenPipeError, ConnectionResetError):
|
||
pass
|
||
|
||
def valid_host(self):
|
||
host = self.headers.get('Host', '')
|
||
if host in (f'127.0.0.1:{args.port}', f'localhost:{args.port}'):
|
||
return True
|
||
if not args.lan:
|
||
return False
|
||
try:
|
||
parsed = urlparse('http://' + host)
|
||
ipaddress.ip_address(parsed.hostname)
|
||
return parsed.port == args.port
|
||
except ValueError:
|
||
return False
|
||
|
||
def authenticated(self):
|
||
return access.allowed(self.client_address[0], self.headers.get('X-VoicePult-Access', ''))
|
||
|
||
def do_GET(self):
|
||
if not self.valid_host():
|
||
return self.send(403, {'error': 'Ungültiger Host.'})
|
||
path = urlparse(self.path).path
|
||
if path.startswith('/api/') and not self.authenticated():
|
||
return self.send(401, {'error': 'Zugangscode erforderlich.'})
|
||
if path == '/api/meter':
|
||
if self.headers.get('X-VoicePult-Token') != token:
|
||
return self.send(403, {'error': 'Zugriff verweigert.'})
|
||
source = parse_qs(urlparse(self.path).query).get('source', [MIC])[0]
|
||
with panel.lock:
|
||
allowed = panel.meter_sources()
|
||
if source not in allowed:
|
||
return self.send(400, {'error': 'Unbekannte Audioquelle.'})
|
||
return self.send(200, meters.snapshot(source))
|
||
if path == '/api/state':
|
||
return self.send(200, {**panel.state(), 'token': token, 'network': access.info(self.client_address[0])})
|
||
files = {'/': ('index.html', 'text/html; charset=utf-8'), '/app.js': ('app.js', 'text/javascript; charset=utf-8'),
|
||
'/style.css': ('style.css', 'text/css; charset=utf-8'), '/icon.svg': ('icon.svg', 'image/svg+xml')}
|
||
if path not in files:
|
||
return self.send(404, {'error': 'Nicht gefunden.'})
|
||
name, mime = files[path]
|
||
self.send(200, (ROOT / 'web' / name).read_bytes(), mime)
|
||
|
||
def do_POST(self):
|
||
if not self.authenticated():
|
||
return self.send(401, {'error': 'Zugangscode erforderlich.'})
|
||
if not self.valid_host() or self.headers.get('X-VoicePult-Token') != token:
|
||
return self.send(403, {'error': 'Zugriff verweigert. Seite neu laden.'})
|
||
if self.path != '/api/action':
|
||
return self.send(404, {'error': 'Nicht gefunden.'})
|
||
try:
|
||
length = int(self.headers.get('Content-Length', '0'))
|
||
if not 0 < length <= 8192:
|
||
raise ValueError('Ungültige Anfragegröße.')
|
||
data = json.loads(self.rfile.read(length))
|
||
if not isinstance(data, dict):
|
||
raise ValueError('Ungültige Anfrage.')
|
||
self.send(200, panel.action(data))
|
||
except (ValueError, TypeError) as exc:
|
||
self.send(400, {'error': str(exc)})
|
||
except AudioError as exc:
|
||
panel.error = str(exc)
|
||
self.send(503, {'error': str(exc)})
|
||
|
||
httpd = ThreadingHTTPServer(('0.0.0.0' if args.lan else '127.0.0.1', args.port), Handler)
|
||
def worker():
|
||
recovered = False
|
||
while not stop.is_set():
|
||
with panel.lock:
|
||
try:
|
||
if not recovered:
|
||
panel.recover()
|
||
recovered = True
|
||
panel.reconcile()
|
||
except (AudioError, ValueError) as exc:
|
||
panel.error = str(exc)
|
||
stop.wait(2)
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
def shutdown(*_):
|
||
stop.set()
|
||
threading.Thread(target=httpd.shutdown, daemon=True).start()
|
||
signal.signal(signal.SIGTERM, shutdown)
|
||
signal.signal(signal.SIGINT, shutdown)
|
||
print(f'VoicePult bereit: http://127.0.0.1:{args.port}', flush=True)
|
||
try:
|
||
httpd.serve_forever()
|
||
finally:
|
||
stop.set()
|
||
httpd.server_close()
|
||
meters.close()
|
||
try:
|
||
panel.close()
|
||
except AudioError as exc:
|
||
print(f'Audio-Bereinigung: {exc}', flush=True)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|