发表于 2026年08月15日
CM101H MPD DAC 管理页面生效
nano /usr/local/bin/mpd-manager.py
#!/usr/bin/env python3
from flask import Flask, request, redirect, render_template_string
import subprocess
import re
import shutil
import os
import html
import time
app = Flask(__name__)
MPD_CONF = "/etc/mpd.conf"
BACKUP_CONF = "/etc/mpd.conf.backup"
HTML = r"""
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport"
content="width=device-width,initial-scale=1">
<title>CM101H MPD DAC管理</title>
<style>
*{
box-sizing:border-box;
}
body{
background:#111;
color:#fff;
font-family:Arial,sans-serif;
margin:0;
padding:20px;
}
.container{
max-width:800px;
margin:auto;
}
h2{
margin-top:0;
}
.box{
background:#222;
padding:20px;
border-radius:14px;
margin-bottom:18px;
box-shadow:0 3px 12px rgba(0,0,0,.35);
}
.card{
background:#181818;
border:1px solid #333;
border-radius:10px;
padding:15px;
margin-bottom:12px;
}
.current{
border:1px solid #4caf50;
}
.title{
font-size:20px;
margin-bottom:10px;
}
.info{
color:#aaa;
line-height:1.7;
}
.ok{
color:#4caf50;
font-weight:bold;
}
.warn{
color:#ff9800;
font-weight:bold;
}
.bad{
color:#f44336;
font-weight:bold;
}
button,
select,
input[type=range]{
width:100%;
}
button,
select{
font-size:17px;
padding:12px;
border-radius:8px;
border:0;
margin-top:8px;
}
button{
background:#444;
color:white;
cursor:pointer;
}
button:hover{
background:#555;
}
select{
background:#333;
color:white;
}
pre{
white-space:pre-wrap;
word-break:break-all;
background:#111;
padding:15px;
border-radius:8px;
overflow:auto;
}
.volume{
font-size:28px;
text-align:center;
margin:15px 0;
}
input[type=range]{
accent-color:#4caf50;
}
.small{
color:#888;
font-size:14px;
}
hr{
border:0;
border-top:1px solid #333;
margin:15px 0;
}
</style>
</head>
<body>
<div class="container">
<h2>CM101H MPD DAC 管理</h2>
<div class="box">
<div class="title">
当前状态
</div>
<div class="info">
当前输出：
<b>{{ current_name }}</b>
<br>
设备：
<b>{{ current_hw }}</b>
<br>
Mixer：
{% if mixer_type == "hardware" %}
<span class="ok">
硬件音量 PCM
</span>
{% elif mixer_type == "software" %}
<span class="ok">
MPD 软件音量
</span>
{% else %}
<span class="warn">
音量控制未知
</span>
{% endif %}
<br>
MPD：
{% if mpd_ok %}
<span class="ok">
运行正常
</span>
{% else %}
<span class="bad">
MPD异常
</span>
{% endif %}
</div>
</div>
<div class="box">
<div class="title">
MPD 音量
</div>
<div class="volume">
{{ volume }}%
</div>
<form method="post"
action="/volume">
<input
type="range"
name="volume"
min="0"
max="100"
value="{{ volume_num }}"
oninput="document.getElementById('vol').innerText=this.value+'%'"
>
<div style="text-align:center;margin-top:10px">
当前：
<span id="vol">
{{ volume }}%
</span>
</div>
<br>
<button type="submit">
设置音量
</button>
</form>
</div>
<div class="box">
<div class="title">
检测到的声卡
</div>
{% if dacs %}
{% for d in dacs %}
<div class="card {% if d.hw == current_hw %}current{% endif %}">
<b>
{{ d.hw }}
</b>
<br>
<div class="info">
名称：
{{ d.name }}
<br>
ALSA：
{{ d.alsa_name }}
<br>
Mixer：
{% if d.mixer == "hardware" %}
<span class="ok">
PCM 硬件音量
</span>
{% else %}
<span class="ok">
MPD 软件音量
</span>
{% endif %}
</div>
<form method="post"
action="/set">
<input
type="hidden"
name="dac"
value="{{ d.hw }}"
>
<button type="submit">
{% if d.hw == current_hw %}
当前正在使用
{% else %}
切换到 {{ d.hw }}
{% endif %}
</button>
</form>
</div>
{% endfor %}
{% else %}
<div class="warn">
没有检测到 ALSA 播放设备
</div>
{% endif %}
</div>
<div class="box">
<div class="title">
MPD 配置
</div>
<pre>{{ current }}</pre>
</div>
<div class="box">
<form method="post"
action="/restart">
<button type="submit">
重启 MPD
</button>
</form>
<br>
<form method="post"
action="/restore">
<button type="submit">
恢复 MPD 原始配置
</button>
</form>
</div>
<div class="small">
CM101H MPD DAC Manager
</div>
</div>
</body>
</html>
"""
# =========================================================
# 基础命令执行
# =========================================================
def run(cmd):
try:
result = subprocess.check_output(
cmd,
shell=True,
stderr=subprocess.STDOUT
)
return result.decode(
errors="ignore"
).strip()
except Exception:
return ""
# =========================================================
# 获取当前 audio_output
# =========================================================
def get_current_output():
try:
text = open(
MPD_CONF,
encoding="utf-8"
).read()
except Exception:
return ""
m = re.search(
r'audio_output\s*\{(.*?)\}',
text,
re.S
)
if not m:
return ""
return m.group(1).strip()
# =========================================================
# 获取当前 DAC
# =========================================================
def get_current_hw():
block = get_current_output()
m = re.search(
r'device\s+"([^"]+)"',
block
)
if m:
return m.group(1)
return ""
def get_current_name():
block = get_current_output()
m = re.search(
r'name\s+"([^"]+)"',
block
)
if m:
return m.group(1)
return ""
# =========================================================
# 获取当前 Mixer
# =========================================================
def get_mixer_type():
block = get_current_output()
m = re.search(
r'mixer_type\s+"([^"]+)"',
block
)
if m:
return m.group(1)
return "none"
# =========================================================
# 获取 MPD 音量
# =========================================================
def get_volume():
out = run("mpc volume")
m = re.search(
r'volume:\s*(\d+)%',
out
)
if m:
return m.group(1)
return "0"
# =========================================================
# MPD 状态
# =========================================================
def get_mpd_status():
result = subprocess.run(
[
"systemctl",
"is-active",
"--quiet",
"mpd"
]
)
return result.returncode == 0
# =========================================================
# 检查 PCM 硬件 Mixer
# =========================================================
def check_hardware_mixer(card):
"""
检测 ALSA 声卡是否存在 PCM 硬件音量。
有 PCM：
hardware
没有 PCM：
software
"""
out = run(
"amixer -c %s scontrols" % card
)
# 最常见：
#
# Simple mixer control 'PCM',0
#
if re.search(
r"Simple mixer control\s+[\"']PCM[\"']",
out,
re.I
):
return "hardware"
# 兼容部分 amixer 输出
if re.search(
r"mixer control\s+[\"']?PCM[\"']?",
out,
re.I
):
return "hardware"
return "software"
# =========================================================
# 获取 PCM Mixer 详细信息
# =========================================================
def get_pcm_info(card):
out = run(
"amixer -c %s get PCM" % card
)
if not out:
return ""
return out
# =========================================================
# 扫描 ALSA DAC
# =========================================================
def scan_dac():
out = run("aplay -l")
result = []
current_card = None
current_card_name = ""
for line in out.splitlines():
card_match = re.search(
r"card\s+(\d+):\s*([^\[]+)\s*\[([^\]]+)\]",
line
)
if card_match:
current_card = card_match.group(1)
current_card_name = (
card_match.group(3).strip()
)
device_match = re.search(
r"device\s+(\d+):\s*([^\[]+)",
line
)
if (
device_match
and current_card is not None
):
device = device_match.group(1)
name = device_match.group(2).strip()
# USB DAC 使用稳定名称
if (
"DAC" in current_card_name.upper()
and
"USB" in current_card_name.upper()
):
hw = "hw:CARD=DAC,DEV=%s" % device
else:
hw = "hw:%s,%s" % (
current_card,
device
)
mixer = check_hardware_mixer(
current_card
)
result.append(
{
"hw": hw,
"name": name,
"alsa_name": current_card_name,
"card": current_card,
"device": device,
"mixer": mixer
}
)
return result
# =========================================================
# 备份 MPD 配置
# =========================================================
def backup_config():
try:
if not os.path.exists(
BACKUP_CONF
):
shutil.copy2(
MPD_CONF,
BACKUP_CONF
)
except Exception:
pass
# =========================================================
# 找到 audio_output 区块
# =========================================================
def find_audio_output(text):
start = text.find(
"audio_output {"
)
if start < 0:
return None, None
count = 0
end = -1
for i in range(
start,
len(text)
):
if text[i] == "{":
count += 1
elif text[i] == "}":
count -= 1
if count == 0:
end = i + 1
break
if end < 0:
return None, None
return start, end
# =========================================================
# 切换 DAC
# =========================================================
def change_dac(hw, name):
if not os.path.exists(MPD_CONF):
return False
# ==============================
# 新版稳定 USB DAC
# ==============================
stable = re.match(
r"^hw:CARD=([^,]+),DEV=(\d+)$",
hw,
re.I
)
if stable:
card_name = stable.group(1).upper()
device = stable.group(2)
if card_name == "DAC":
mixer_test = run(
"amixer -D hw:DAC scontrols"
)
if re.search(
r"Simple mixer control\s+[\"']PCM[\"']",
mixer_test,
re.I
):
mixer_config = """
mixer_type "hardware"
mixer_device "hw:DAC"
mixer_control "PCM"
"""
else:
mixer_config = """
mixer_type "software"
"""
new = """
audio_output {
type "alsa"
name "%s"
device "hw:CARD=DAC,DEV=%s"
%s
auto_resample "no"
auto_channels "no"
auto_format "no"
}
""" % (
name,
device,
mixer_config
)
else:
return False
else:
# 兼容旧 hw:1,0
m = re.match(
r"^hw:(\d+),(\d+)$",
hw
)
if not m:
return False
card = m.group(1)
device = m.group(2)
mixer = check_hardware_mixer(card)
if mixer == "hardware":
mixer_config = """
mixer_type "hardware"
mixer_device "hw:%s"
mixer_control "PCM"
""" % card
else:
mixer_config = """
mixer_type "software"
"""
new = """
audio_output {
type "alsa"
name "%s"
device "hw:%s,%s"
%s
auto_resample "no"
auto_channels "no"
auto_format "no"
}
""" % (
name,
card,
device,
mixer_config
)
try:
text = open(
MPD_CONF,
encoding="utf-8"
).read()
except Exception:
return False
backup_config()
start,end = find_audio_output(text)
if start is None:
return False
new_text = (
text[:start]
+
new
+
text[end:]
)
try:
with open(
MPD_CONF,
"w",
encoding="utf-8"
) as f:
f.write(new_text)
except Exception:
return False
return True
# =========================================================
# 重启 MPD
# =========================================================
def restart_mpd():
try:
subprocess.run(
[
"systemctl",
"restart",
"mpd"
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=20
)
return True
except Exception:
return False
# =========================================================
# 首页
# =========================================================
@app.route("/")
def index():
volume = get_volume()
try:
volume_num = int(
volume
)
except Exception:
volume_num = 0
current_config = run(
"grep -A20 'audio_output {' /etc/mpd.conf"
)
return render_template_string(
HTML,
dacs=scan_dac(),
current=current_config,
current_hw=get_current_hw(),
current_name=get_current_name(),
mixer_type=get_mixer_type(),
volume=volume,
volume_num=volume_num,
mpd_ok=get_mpd_status()
)
# =========================================================
# DAC 切换
# =========================================================
@app.route(
"/set",
methods=["POST"]
)
def setdac():
hw = request.form.get(
"dac",
""
)
# 安全检查
#
# 支持：
# hw:0,0
# hw:1,0
# hw:CARD=DAC,DEV=0
#
if not (
re.match(
r"^hw:\d+,\d+$",
hw
)
or
re.match(
r"^hw:CARD=[^,]+,DEV=\d+$",
hw,
re.I
)
):
return redirect("/")
# -----------------------------------------------------
# 从当前 ALSA 列表取得真实名称
# -----------------------------------------------------
name = "USB DAC"
for d in scan_dac():
if d["hw"] == hw:
name = d["name"]
break
# -----------------------------------------------------
# 切换
# -----------------------------------------------------
success = change_dac(
hw,
name
)
if success:
restart_mpd()
# 等待 MPD 完成启动
time.sleep(1)
return redirect("/")
# =========================================================
# 设置音量
# =========================================================
@app.route(
"/volume",
methods=["POST"]
)
def volume():
value = request.form.get(
"volume",
"50"
)
try:
value = int(
value
)
except Exception:
value = 50
value = max(
0,
min(
100,
value
)
)
# -----------------------------------------------------
# MPD 统一通过 mpc volume 设置
#
# hardware：
# MPD -> ALSA PCM Mixer
#
# software：
# MPD 软件音量
# -----------------------------------------------------
subprocess.run(
[
"mpc",
"volume",
str(value)
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
return redirect("/")
# =========================================================
# 手动重启 MPD
# =========================================================
@app.route(
"/restart",
methods=["POST"]
)
def restart():
restart_mpd()
time.sleep(1)
return redirect("/")
# =========================================================
# 恢复原始配置
# =========================================================
@app.route(
"/restore",
methods=["POST"]
)
def restore():
if os.path.exists(
BACKUP_CONF
):
try:
shutil.copy2(
BACKUP_CONF,
MPD_CONF
)
restart_mpd()
time.sleep(1)
except Exception:
pass
return redirect("/")
# =========================================================
# 启动
# =========================================================
if __name__ == "__main__":
app.run(
host="0.0.0.0",
port=8090,
threaded=True
)
python3 -m py_compile /usr/local/bin/mpd-manager.py
systemctl restart mpd-manager
systemctl list-units --type=service | grep -i mpd