DGM20 control for Linux: protocol library, CLI and local web UI
Reverse-engineered from Maono Link 4.0.88 (DGM20_OLD.dll) and verified against a DGM20 (352f:0105, firmware 2.1.2). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1 @@
|
||||
"""Linux control for the Maono DGM20 USB microphone."""
|
||||
@@ -0,0 +1,87 @@
|
||||
"""dgm20ctl: command-line control for the Maono DGM20."""
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from .device import SETTINGS, DeviceError, Mic, describe
|
||||
|
||||
|
||||
def fmt(value):
|
||||
if value is None:
|
||||
return "?"
|
||||
if isinstance(value, bool):
|
||||
return "on" if value else "off"
|
||||
return str(value)
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
ap = argparse.ArgumentParser(
|
||||
prog="dgm20ctl",
|
||||
description="Control Maono DGM20 microphone settings.",
|
||||
epilog="settings: " + ", ".join(f"{n} ({describe(e)})" for n, (_, e, _) in SETTINGS.items()))
|
||||
ap.add_argument("--dev", help="hidraw node (default: autodetect)")
|
||||
sub = ap.add_subparsers(dest="cmd")
|
||||
sub.add_parser("status", help="show current settings (default)")
|
||||
sub.add_parser("info", help="model and firmware version")
|
||||
sub.add_parser("watch", help="print changes made on the mic itself")
|
||||
s = sub.add_parser("set", help="change settings, e.g. set nc on nc-level 80")
|
||||
s.add_argument("pairs", nargs="+", metavar="NAME VALUE")
|
||||
r = sub.add_parser("factory-reset", help="restore factory defaults")
|
||||
r.add_argument("--yes", action="store_true", help="confirm the reset")
|
||||
raw = sub.add_parser("raw", help="raw register access, for exploring")
|
||||
raw.add_argument("op", choices=["get", "set", "dump"])
|
||||
raw.add_argument("addr", type=lambda v: int(v, 0))
|
||||
raw.add_argument("value", type=lambda v: int(v, 0), nargs="?", help="value (set) or count (dump)")
|
||||
a = ap.parse_args(argv)
|
||||
|
||||
if a.cmd == "set":
|
||||
if len(a.pairs) % 2:
|
||||
ap.error("set takes NAME VALUE pairs")
|
||||
for name in a.pairs[0::2]:
|
||||
if name not in SETTINGS:
|
||||
ap.error(f"unknown setting {name!r}; known: {', '.join(SETTINGS)}")
|
||||
if a.cmd == "factory-reset" and not a.yes:
|
||||
ap.error("factory-reset resets every setting on the mic; rerun with --yes")
|
||||
|
||||
try:
|
||||
with Mic(a.dev) as mic:
|
||||
run(mic, a)
|
||||
except (DeviceError, ValueError) as e:
|
||||
sys.exit(f"dgm20ctl: {e}")
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
|
||||
def run(mic, a):
|
||||
if a.cmd in (None, "status"):
|
||||
for name, value in mic.state().items():
|
||||
print(f"{name:11} {fmt(value):>5} {SETTINGS[name][2]}")
|
||||
elif a.cmd == "info":
|
||||
for k, v in mic.info().items():
|
||||
print(f"{k + ':':9} {v or '?'}")
|
||||
elif a.cmd == "watch":
|
||||
print("watching for changes made on the mic, ctrl-c to stop")
|
||||
for name, value in mic.events():
|
||||
print(f"{name} = {fmt(value)}", flush=True)
|
||||
elif a.cmd == "set":
|
||||
for name, value in zip(a.pairs[0::2], a.pairs[1::2]):
|
||||
print(f"{name} = {fmt(mic.set(name, value))}")
|
||||
elif a.cmd == "factory-reset":
|
||||
print("settings restored to factory defaults" if mic.factory_reset()
|
||||
else "the mic rejected the reset")
|
||||
elif a.op == "get":
|
||||
v = mic.read(a.addr)
|
||||
print("no reply" if v is None else v)
|
||||
elif a.op == "set":
|
||||
if a.value is None:
|
||||
raise ValueError("raw set needs a value")
|
||||
print(f"{a.addr:#06x} = {mic.read(a.addr)}" if mic.write(a.addr, a.value) else "rejected")
|
||||
else:
|
||||
vals = mic.read_range(a.addr, a.value or 1)
|
||||
if vals is None:
|
||||
raise DeviceError("no reply")
|
||||
for i, v in enumerate(vals):
|
||||
print(f"{a.addr + i:#06x} {v:5} {v:#06x}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,220 @@
|
||||
"""Maono DGM20 vendor HID protocol.
|
||||
|
||||
The mic exposes a vendor HID channel next to its USB audio interfaces. The
|
||||
host writes output report 0x51 and the mic answers with input report 0x11.
|
||||
Both share one framing:
|
||||
|
||||
[0] report id (0x51 out / 0x11 in)
|
||||
[1] 0xC4 magic
|
||||
[2] length of bytes [1..] including the checksum
|
||||
[3-4] 00 00
|
||||
[5] type: 2 = write ack (payload u16 status, 0 = ok)
|
||||
3 = write (addr,val) pairs; the mic also sends these when
|
||||
a setting changes on the hardware
|
||||
4 = read one register
|
||||
5 = bulk write
|
||||
6 = read a range (addr, count)
|
||||
[6..] payload, u16 little-endian
|
||||
[-2:] checksum u16 LE = -(sum of bytes [1..end-2])
|
||||
|
||||
Settings are 16-bit registers. Input gain and headphone volume are standard
|
||||
USB audio controls and are not part of this channel.
|
||||
"""
|
||||
import glob
|
||||
import os
|
||||
import select
|
||||
import threading
|
||||
|
||||
VID, PID = 0x352F, 0x0105
|
||||
REPORT_OUT, REPORT_IN, MAGIC = 0x51, 0x11, 0xC4
|
||||
T_ACK, T_WRITE, T_READ, T_BULK, T_RANGE = 2, 3, 4, 5, 6
|
||||
|
||||
# name: (register, encoding, description)
|
||||
# encodings: "bool" = 0/1, (lo, hi) = integer range, dict = word -> raw value
|
||||
SETTINGS = {
|
||||
"mute": (0x2022, "bool", "Mute the mic (same as the touch pad)"),
|
||||
"nc": (0x2027, "bool", "Noise cancellation"),
|
||||
"nc-level": (0x2028, (0, 100), "Noise cancellation strength"),
|
||||
"enhance": (0x202C, "bool", "Vocal enhancement"),
|
||||
"monitor": (0x2026, {"on": 2000, "off": 3000}, "Hear yourself in headphones"),
|
||||
"brightness": (0x202A, (0, 100), "RGB ring brightness"),
|
||||
}
|
||||
REG_FACTORY_RESET = 0x100A
|
||||
REG_DFU = 0x1008
|
||||
REG_MODEL, REG_VERSION = 0x0003, 0x000B
|
||||
|
||||
|
||||
class DeviceError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def find_hidraw():
|
||||
for node in sorted(glob.glob("/sys/class/hidraw/hidraw*")):
|
||||
try:
|
||||
with open(os.path.join(node, "device/uevent")) as f:
|
||||
uevent = f.read()
|
||||
except OSError:
|
||||
continue
|
||||
if f"HID_ID=0003:{VID:08X}:{PID:08X}" in uevent:
|
||||
return "/dev/" + os.path.basename(node)
|
||||
return None
|
||||
|
||||
|
||||
def frame(typ, words):
|
||||
body = bytearray([MAGIC, 0, 0, 0, typ])
|
||||
for w in words:
|
||||
body += bytes([w & 0xFF, (w >> 8) & 0xFF])
|
||||
body[1] = len(body) + 2
|
||||
ck = -sum(body) & 0xFFFF
|
||||
body += bytes([ck & 0xFF, ck >> 8])
|
||||
return bytes([REPORT_OUT]) + bytes(body) + bytes(64 - len(body))
|
||||
|
||||
|
||||
def parse(r):
|
||||
"""Return (type, [u16 words]) for a well-formed reply, else None."""
|
||||
if len(r) < 8 or r[0] != REPORT_IN or r[1] != MAGIC:
|
||||
return None
|
||||
n = r[2]
|
||||
if n < 7 or n + 1 > len(r) or (sum(r[1:n - 1]) + (r[n - 1] | r[n] << 8)) & 0xFFFF:
|
||||
return None
|
||||
payload = r[6:n - 1]
|
||||
return r[5], [payload[i] | payload[i + 1] << 8 for i in range(0, len(payload) - 1, 2)]
|
||||
|
||||
|
||||
def decode(name, raw):
|
||||
"""Raw register value -> friendly value (bool, int or str)."""
|
||||
enc = SETTINGS[name][1]
|
||||
if raw is None:
|
||||
return None
|
||||
if enc == "bool":
|
||||
return bool(raw)
|
||||
if isinstance(enc, dict):
|
||||
return next((k for k, v in enc.items() if v == raw), raw)
|
||||
return raw
|
||||
|
||||
|
||||
def encode(name, value):
|
||||
"""Friendly value (bool, int, or 'on'/'off'/number string) -> raw register value."""
|
||||
if name not in SETTINGS:
|
||||
raise ValueError(f"unknown setting {name!r}; known: {', '.join(SETTINGS)}")
|
||||
enc = SETTINGS[name][1]
|
||||
if isinstance(enc, tuple):
|
||||
if isinstance(value, str) and value.strip().isdigit():
|
||||
value = int(value)
|
||||
if isinstance(value, int) and not isinstance(value, bool) and enc[0] <= value <= enc[1]:
|
||||
return value
|
||||
else:
|
||||
if isinstance(value, str):
|
||||
value = {"on": True, "true": True, "1": True,
|
||||
"off": False, "false": False, "0": False}.get(value.strip().lower(), value)
|
||||
if isinstance(value, bool):
|
||||
return enc["on" if value else "off"] if isinstance(enc, dict) else int(value)
|
||||
raise ValueError(f"{name}: expected {describe(enc)}, got {value!r}")
|
||||
|
||||
|
||||
def describe(enc):
|
||||
if isinstance(enc, tuple):
|
||||
return f"{enc[0]}-{enc[1]}"
|
||||
return "on|off"
|
||||
|
||||
|
||||
def words_to_ascii(words):
|
||||
return bytes(b for w in words for b in (w & 0xFF, w >> 8)).split(b"\0")[0].decode(errors="replace")
|
||||
|
||||
|
||||
class Mic:
|
||||
"""One open handle to the mic. Thread-safe: each transaction holds a lock."""
|
||||
|
||||
def __init__(self, path=None):
|
||||
self.path = path or find_hidraw()
|
||||
if not self.path:
|
||||
raise DeviceError("DGM20 (352f:0105) not found")
|
||||
try:
|
||||
self.fd = os.open(self.path, os.O_RDWR)
|
||||
except PermissionError:
|
||||
raise DeviceError(f"no permission to open {self.path}; install the udev rule") from None
|
||||
except OSError as e:
|
||||
raise DeviceError(f"cannot open {self.path}: {e.strerror}") from None
|
||||
self.lock = threading.Lock()
|
||||
self._drain()
|
||||
|
||||
def close(self):
|
||||
os.close(self.fd)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc):
|
||||
self.close()
|
||||
|
||||
def _drain(self):
|
||||
while select.select([self.fd], [], [], 0)[0]:
|
||||
os.read(self.fd, 64)
|
||||
|
||||
def _io(self, typ, words, accept, timeout=0.5):
|
||||
with self.lock:
|
||||
try:
|
||||
self._drain()
|
||||
os.write(self.fd, frame(typ, words))
|
||||
while select.select([self.fd], [], [], timeout)[0]:
|
||||
p = parse(os.read(self.fd, 64))
|
||||
if p and accept(p):
|
||||
return p[1]
|
||||
except OSError as e:
|
||||
raise DeviceError(f"mic disconnected ({e.strerror})") from None
|
||||
return None
|
||||
|
||||
def read(self, addr):
|
||||
w = self._io(T_READ, [addr], lambda p: p[0] == T_READ and p[1][:1] == [addr])
|
||||
return None if w is None else w[1]
|
||||
|
||||
def read_range(self, addr, count):
|
||||
w = self._io(T_RANGE, [addr, count], lambda p: p[0] == T_RANGE and p[1][:1] == [addr])
|
||||
return None if w is None else w[2:2 + count]
|
||||
|
||||
def write(self, addr, value):
|
||||
w = self._io(T_WRITE, [addr, value], lambda p: p[0] == T_ACK)
|
||||
return w == [0]
|
||||
|
||||
# friendly layer
|
||||
|
||||
def get(self, name):
|
||||
return decode(name, self.read(SETTINGS[name][0]))
|
||||
|
||||
def set(self, name, value):
|
||||
raw = encode(name, value)
|
||||
if not self.write(SETTINGS[name][0], raw):
|
||||
raise DeviceError(f"mic rejected {name} = {value}")
|
||||
return self.get(name)
|
||||
|
||||
def state(self):
|
||||
return {name: self.get(name) for name in SETTINGS}
|
||||
|
||||
def info(self):
|
||||
model = self.read_range(REG_MODEL, 8)
|
||||
ver = self.read_range(REG_VERSION, 3)
|
||||
return {
|
||||
"model": words_to_ascii(model) if model else None,
|
||||
"firmware": ".".join(map(str, ver)) if ver else None,
|
||||
"path": self.path,
|
||||
}
|
||||
|
||||
def factory_reset(self):
|
||||
return self.write(REG_FACTORY_RESET, 1)
|
||||
|
||||
def events(self):
|
||||
"""Yield (name_or_addr, value) for changes the mic reports on its own."""
|
||||
by_addr = {a: n for n, (a, _, _) in SETTINGS.items()}
|
||||
while True:
|
||||
select.select([self.fd], [], [])
|
||||
with self.lock:
|
||||
try:
|
||||
p = parse(os.read(self.fd, 64))
|
||||
except OSError as e:
|
||||
raise DeviceError(f"mic disconnected ({e.strerror})") from None
|
||||
if not p or p[0] != T_WRITE:
|
||||
continue
|
||||
w = p[1]
|
||||
for addr, val in zip(w[0::2], w[1::2]):
|
||||
name = by_addr.get(addr)
|
||||
yield (name, decode(name, val)) if name else (f"{addr:#06x}", val)
|
||||
@@ -0,0 +1,337 @@
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>DGM20 Control</title>
|
||||
<link rel="icon" href="data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 32 32'%3E%3Ccircle cx='16' cy='16' r='13' fill='none' stroke='%232E7D5B' stroke-width='4'/%3E%3Ccircle cx='16' cy='16' r='6' fill='%231D232B'/%3E%3C/svg%3E">
|
||||
<style>
|
||||
:root {
|
||||
--base: #E4E6E3;
|
||||
--panel: #F4F5F2;
|
||||
--ink: #1D232B;
|
||||
--ink-2: #5B6470;
|
||||
--rule: #CFD3CF;
|
||||
--live: #2E7D5B;
|
||||
--muted: #C8442B;
|
||||
--off: #9AA1A8;
|
||||
--grille: #2A3038;
|
||||
--grille-dot: #3A424C;
|
||||
--focus: #1F6FD1;
|
||||
--ring-glow: 1;
|
||||
}
|
||||
@media (prefers-color-scheme: dark) {
|
||||
:root {
|
||||
--base: #16191D;
|
||||
--panel: #1F2327;
|
||||
--ink: #E8EAEC;
|
||||
--ink-2: #9AA3AD;
|
||||
--rule: #30363C;
|
||||
--live: #4FB386;
|
||||
--muted: #E5634A;
|
||||
--off: #59616A;
|
||||
--grille: #0F1215;
|
||||
--grille-dot: #262C33;
|
||||
--focus: #6AA8FF;
|
||||
}
|
||||
}
|
||||
* { box-sizing: border-box; }
|
||||
html { -webkit-text-size-adjust: 100%; }
|
||||
body {
|
||||
margin: 0;
|
||||
min-height: 100vh;
|
||||
background: var(--base);
|
||||
color: var(--ink);
|
||||
font: 16px/1.5 system-ui, -apple-system, "Segoe UI", Cantarell, "Noto Sans", sans-serif;
|
||||
display: flex;
|
||||
justify-content: center;
|
||||
padding: 32px 16px 48px;
|
||||
}
|
||||
.shell { width: 100%; max-width: 460px; }
|
||||
|
||||
header { display: flex; align-items: baseline; justify-content: space-between; gap: 12px; margin-bottom: 28px; }
|
||||
.wordmark { font-size: 1.375rem; font-weight: 700; letter-spacing: -0.01em; margin: 0; }
|
||||
.wordmark span { font-weight: 400; color: var(--ink-2); }
|
||||
.conn { font-size: 0.8125rem; color: var(--ink-2); display: flex; align-items: center; gap: 6px; }
|
||||
.conn::before { content: ""; width: 8px; height: 8px; border-radius: 50%; background: var(--off); }
|
||||
.conn[data-on="1"]::before { background: var(--live); }
|
||||
|
||||
/* The mute pad: the mic's capacitive top, seen from above. */
|
||||
.pad-wrap { display: flex; flex-direction: column; align-items: center; margin-bottom: 32px; }
|
||||
.pad {
|
||||
--state: var(--live);
|
||||
position: relative;
|
||||
width: min(232px, 64vw);
|
||||
aspect-ratio: 1;
|
||||
border-radius: 50%;
|
||||
border: 0;
|
||||
padding: 0;
|
||||
cursor: pointer;
|
||||
background:
|
||||
radial-gradient(circle at 50% 50%, var(--grille-dot) 1.6px, transparent 2.2px) 0 0 / 11px 11px,
|
||||
var(--grille);
|
||||
box-shadow:
|
||||
0 0 0 6px var(--panel),
|
||||
0 0 0 10px color-mix(in srgb, var(--state) calc(var(--ring-glow) * 100%), var(--rule)),
|
||||
0 0 calc(var(--ring-glow) * 46px) calc(var(--ring-glow) * 6px) color-mix(in srgb, var(--state) 45%, transparent);
|
||||
transition: box-shadow 220ms ease;
|
||||
}
|
||||
.pad[aria-pressed="true"] { --state: var(--muted); }
|
||||
.pad:disabled { --state: var(--off); --ring-glow: 0; cursor: default; }
|
||||
.pad:focus-visible { outline: 3px solid var(--focus); outline-offset: 18px; }
|
||||
.pad svg { position: absolute; inset: 0; margin: auto; width: 34%; height: 34%; color: #F4F5F2; opacity: .92; }
|
||||
.pad .slash { opacity: 0; transition: opacity 160ms ease; }
|
||||
.pad[aria-pressed="true"] .slash { opacity: 1; }
|
||||
.pad-state { margin: 26px 0 2px; font-size: 1.25rem; font-weight: 650; color: var(--state-ink, var(--live)); }
|
||||
.pad-state[data-s="muted"] { --state-ink: var(--muted); }
|
||||
.pad-state[data-s="none"] { --state-ink: var(--ink-2); }
|
||||
.pad-hint { font-size: 0.875rem; color: var(--ink-2); margin: 0; }
|
||||
|
||||
/* Settings list */
|
||||
.settings { background: var(--panel); border-radius: 14px; padding: 4px 20px; margin: 0; list-style: none; }
|
||||
.row { display: grid; grid-template-columns: 1fr auto; gap: 4px 16px; align-items: center; padding: 16px 0; }
|
||||
.row + .row { border-top: 1px solid var(--rule); }
|
||||
.row.sub { border-top: 0; padding-top: 0; }
|
||||
.row h2 { font-size: 1rem; font-weight: 600; margin: 0; }
|
||||
.row p { grid-column: 1; font-size: 0.875rem; color: var(--ink-2); margin: 0; }
|
||||
.row .ctl { grid-column: 2; grid-row: 1 / span 2; }
|
||||
.row.slider { grid-template-columns: auto 1fr 3ch; gap: 14px; }
|
||||
.row.slider label { font-size: 0.9375rem; color: var(--ink-2); }
|
||||
.row.slider output { font-variant-numeric: tabular-nums; text-align: right; font-weight: 600; }
|
||||
.row[data-disabled="1"] { opacity: .45; }
|
||||
|
||||
.switch {
|
||||
width: 48px; height: 28px; border-radius: 14px; border: 0; padding: 3px; cursor: pointer;
|
||||
background: var(--off); display: flex; transition: background 160ms ease;
|
||||
}
|
||||
.switch::after {
|
||||
content: ""; width: 22px; height: 22px; border-radius: 50%; background: #fff;
|
||||
box-shadow: 0 1px 2px rgba(0,0,0,.25); transition: transform 160ms ease;
|
||||
}
|
||||
.switch[aria-checked="true"] { background: var(--live); }
|
||||
.switch[aria-checked="true"]::after { transform: translateX(20px); }
|
||||
.switch:focus-visible, input[type=range]:focus-visible, .link:focus-visible { outline: 3px solid var(--focus); outline-offset: 3px; }
|
||||
.switch:disabled { cursor: default; }
|
||||
|
||||
input[type=range] { width: 100%; accent-color: var(--live); margin: 0; }
|
||||
|
||||
.notice {
|
||||
background: var(--panel); border-left: 4px solid var(--muted); border-radius: 8px;
|
||||
padding: 14px 16px; margin-bottom: 20px; font-size: 0.9375rem;
|
||||
}
|
||||
.notice[hidden] { display: none; }
|
||||
.notice strong { display: block; margin-bottom: 2px; }
|
||||
.notice code { font-size: 0.8125rem; word-break: break-all; }
|
||||
|
||||
footer { margin-top: 28px; display: flex; justify-content: space-between; flex-wrap: wrap; gap: 8px 16px; font-size: 0.8125rem; color: var(--ink-2); }
|
||||
.link { background: none; border: 0; padding: 0; font: inherit; color: var(--ink-2); text-decoration: underline; text-underline-offset: 3px; cursor: pointer; }
|
||||
.link.danger { color: var(--muted); }
|
||||
.reset-confirm { display: flex; gap: 14px; align-items: baseline; }
|
||||
.reset-confirm[hidden] { display: none; }
|
||||
|
||||
@media (prefers-reduced-motion: reduce) {
|
||||
*, *::before, *::after { transition: none !important; }
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="shell">
|
||||
<header>
|
||||
<h1 class="wordmark">DGM20 <span>control</span></h1>
|
||||
<div class="conn" id="conn" data-on="0">Connecting</div>
|
||||
</header>
|
||||
|
||||
<div class="notice" id="notice" hidden>
|
||||
<strong id="notice-title"></strong>
|
||||
<span id="notice-body"></span>
|
||||
</div>
|
||||
|
||||
<div class="pad-wrap">
|
||||
<button class="pad" id="pad" aria-pressed="false" aria-label="Mute microphone" disabled>
|
||||
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true">
|
||||
<rect x="9" y="3" width="6" height="11" rx="3"/>
|
||||
<path d="M5.5 11a6.5 6.5 0 0 0 13 0M12 17.5V21M8.5 21h7"/>
|
||||
<path class="slash" d="M4 4l16 16" stroke-width="2.2"/>
|
||||
</svg>
|
||||
</button>
|
||||
<p class="pad-state" id="pad-state" data-s="none">Waiting for mic</p>
|
||||
<p class="pad-hint" id="pad-hint"> </p>
|
||||
</div>
|
||||
|
||||
<ul class="settings">
|
||||
<li class="row">
|
||||
<h2 id="l-nc">Noise cancellation</h2>
|
||||
<p>Filters out keyboard clatter, fans and room hum.</p>
|
||||
<button class="switch ctl" role="switch" aria-checked="false" aria-labelledby="l-nc" data-name="nc" disabled></button>
|
||||
</li>
|
||||
<li class="row slider sub" id="nc-level-row">
|
||||
<label for="nc-level">Strength</label>
|
||||
<input type="range" id="nc-level" min="0" max="100" step="1" data-name="nc-level" disabled>
|
||||
<output for="nc-level" id="nc-level-out">–</output>
|
||||
</li>
|
||||
<li class="row">
|
||||
<h2 id="l-enhance">Vocal enhancement</h2>
|
||||
<p>Brings your voice forward over background sound.</p>
|
||||
<button class="switch ctl" role="switch" aria-checked="false" aria-labelledby="l-enhance" data-name="enhance" disabled></button>
|
||||
</li>
|
||||
<li class="row">
|
||||
<h2 id="l-monitor">Hear yourself</h2>
|
||||
<p>Plays the mic through headphones plugged into the DGM20.</p>
|
||||
<button class="switch ctl" role="switch" aria-checked="false" aria-labelledby="l-monitor" data-name="monitor" disabled></button>
|
||||
</li>
|
||||
<li class="row slider">
|
||||
<label for="brightness">Ring brightness</label>
|
||||
<input type="range" id="brightness" min="0" max="100" step="1" data-name="brightness" disabled>
|
||||
<output for="brightness" id="brightness-out">–</output>
|
||||
</li>
|
||||
</ul>
|
||||
|
||||
<footer>
|
||||
<span id="fw">Unofficial tool, not affiliated with Maono.</span>
|
||||
<span>
|
||||
<button class="link" id="reset">Restore factory settings</button>
|
||||
<span class="reset-confirm" id="reset-confirm" hidden>
|
||||
<span>Reset every setting?</span>
|
||||
<button class="link danger" id="reset-yes">Reset</button>
|
||||
<button class="link" id="reset-no">Cancel</button>
|
||||
</span>
|
||||
</span>
|
||||
</footer>
|
||||
</div>
|
||||
|
||||
<script>
|
||||
"use strict";
|
||||
const $ = (id) => document.getElementById(id);
|
||||
const pad = $("pad");
|
||||
const switches = [...document.querySelectorAll(".switch")];
|
||||
const sliders = [...document.querySelectorAll("input[type=range]")];
|
||||
let state = null;
|
||||
let busy = new Set(); // settings with a write in flight or a drag in progress
|
||||
let lastConnected = null;
|
||||
|
||||
async function api(path, body) {
|
||||
const opts = body === undefined ? {} : {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify(body),
|
||||
};
|
||||
const r = await fetch(path, opts);
|
||||
const data = await r.json().catch(() => ({}));
|
||||
if (!r.ok && !("connected" in data)) throw new Error(data.error || `HTTP ${r.status}`);
|
||||
return data;
|
||||
}
|
||||
|
||||
// body is plain text; `code` (optional) is shown after it in a <code> element
|
||||
function notice(title, body, code) {
|
||||
$("notice").hidden = !title;
|
||||
$("notice-title").textContent = title || "";
|
||||
const b = $("notice-body");
|
||||
b.textContent = body || "";
|
||||
if (code) { const c = document.createElement("code"); c.textContent = code; b.append(" ", c); }
|
||||
}
|
||||
|
||||
function render(snap) {
|
||||
const on = !!snap.connected;
|
||||
$("conn").dataset.on = on ? "1" : "0";
|
||||
$("conn").textContent = on ? "Connected" : "Not connected";
|
||||
pad.disabled = !on;
|
||||
switches.forEach((s) => (s.disabled = !on));
|
||||
sliders.forEach((s) => (s.disabled = !on));
|
||||
|
||||
if (!on) {
|
||||
const err = snap.error || "";
|
||||
if (/permission/i.test(err)) {
|
||||
notice("The mic is plugged in, but this page can't talk to it.",
|
||||
"Install the udev rule, then unplug and replug the mic:", "sudo cp udev/70-maono-dgm20.rules /etc/udev/rules.d/");
|
||||
} else {
|
||||
notice("Mic not found.", "Plug in the DGM20. This page reconnects on its own.");
|
||||
}
|
||||
$("pad-state").textContent = "Waiting for mic";
|
||||
$("pad-state").dataset.s = "none";
|
||||
$("pad-hint").textContent = "\u00a0";
|
||||
lastConnected = false;
|
||||
return;
|
||||
}
|
||||
notice(null);
|
||||
state = snap.state;
|
||||
if (snap.info && snap.info.firmware) {
|
||||
$("fw").textContent = `Firmware ${snap.info.firmware}. Unofficial tool, not affiliated with Maono.`;
|
||||
}
|
||||
|
||||
if (!busy.has("mute")) {
|
||||
pad.setAttribute("aria-pressed", String(!!state.mute));
|
||||
pad.setAttribute("aria-label", state.mute ? "Unmute microphone" : "Mute microphone");
|
||||
$("pad-state").textContent = state.mute ? "Muted" : "Live";
|
||||
$("pad-state").dataset.s = state.mute ? "muted" : "live";
|
||||
$("pad-hint").textContent = state.mute ? "Click to unmute" : "Click to mute";
|
||||
}
|
||||
for (const s of switches) {
|
||||
const n = s.dataset.name;
|
||||
if (busy.has(n)) continue;
|
||||
const v = state[n];
|
||||
s.setAttribute("aria-checked", String(v === true || v === "on"));
|
||||
}
|
||||
for (const s of sliders) {
|
||||
const n = s.dataset.name;
|
||||
if (busy.has(n) || state[n] == null) continue;
|
||||
s.value = state[n];
|
||||
$(`${n}-out`).textContent = state[n];
|
||||
}
|
||||
$("nc-level-row").dataset.disabled = state.nc ? "0" : "1";
|
||||
document.documentElement.style.setProperty("--ring-glow", ((state.brightness ?? 100) / 100 * 0.85 + 0.15).toFixed(2));
|
||||
lastConnected = true;
|
||||
}
|
||||
|
||||
async function poll() {
|
||||
try { render(await api("/api/state")); }
|
||||
catch (e) { render({ connected: false, error: "server unreachable" }); notice("This page lost its server.", "Start it again with", "dgm20-web"); }
|
||||
}
|
||||
|
||||
async function setValue(name, value) {
|
||||
busy.add(name);
|
||||
try { render(await api("/api/set", { name, value })); }
|
||||
catch (e) { notice(`Couldn't change ${name}.`, e.message); }
|
||||
finally { busy.delete(name); }
|
||||
}
|
||||
|
||||
pad.addEventListener("click", () => {
|
||||
const next = pad.getAttribute("aria-pressed") !== "true";
|
||||
pad.setAttribute("aria-pressed", String(next));
|
||||
setValue("mute", next);
|
||||
});
|
||||
|
||||
for (const s of switches) {
|
||||
s.addEventListener("click", () => {
|
||||
const next = s.getAttribute("aria-checked") !== "true";
|
||||
s.setAttribute("aria-checked", String(next));
|
||||
setValue(s.dataset.name, next);
|
||||
});
|
||||
}
|
||||
|
||||
for (const s of sliders) {
|
||||
const n = s.dataset.name;
|
||||
let timer = null;
|
||||
s.addEventListener("input", () => {
|
||||
busy.add(n);
|
||||
$(`${n}-out`).textContent = s.value;
|
||||
if (n === "brightness") document.documentElement.style.setProperty("--ring-glow", (s.value / 100 * 0.85 + 0.15).toFixed(2));
|
||||
clearTimeout(timer);
|
||||
timer = setTimeout(() => api("/api/set", { name: n, value: Number(s.value) }).catch(() => {}), 120);
|
||||
});
|
||||
s.addEventListener("change", () => { clearTimeout(timer); setValue(n, Number(s.value)); });
|
||||
}
|
||||
|
||||
$("reset").addEventListener("click", () => { $("reset").hidden = true; $("reset-confirm").hidden = false; $("reset-yes").focus(); });
|
||||
$("reset-no").addEventListener("click", () => { $("reset").hidden = false; $("reset-confirm").hidden = true; });
|
||||
$("reset-yes").addEventListener("click", async () => {
|
||||
$("reset-confirm").hidden = true; $("reset").hidden = false;
|
||||
try { await api("/api/factory-reset", { confirm: true }); setTimeout(poll, 2500); }
|
||||
catch (e) { notice("The reset didn't go through.", e.message); }
|
||||
});
|
||||
|
||||
poll();
|
||||
setInterval(() => { if (!document.hidden) poll(); }, 1500);
|
||||
document.addEventListener("visibilitychange", () => { if (!document.hidden) poll(); });
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,175 @@
|
||||
"""dgm20-web: a small local web UI for the Maono DGM20.
|
||||
|
||||
Serves one page plus a JSON API. Binds to 127.0.0.1 unless told otherwise.
|
||||
Requests from other sites are refused: the Host header must match, and API
|
||||
writes must be application/json, which browsers won't send cross-origin
|
||||
without a CORS preflight that this server never approves.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import threading
|
||||
import webbrowser
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from importlib.resources import files
|
||||
|
||||
from .device import SETTINGS, DeviceError, Mic, describe
|
||||
|
||||
INDEX = files("dgm20").joinpath("static/index.html").read_bytes()
|
||||
|
||||
|
||||
class MicHolder:
|
||||
"""Keeps one Mic open and reopens it after an unplug."""
|
||||
|
||||
def __init__(self, path=None):
|
||||
self.path, self.mic, self.info, self.lock = path, None, None, threading.Lock()
|
||||
|
||||
def call(self, fn):
|
||||
with self.lock:
|
||||
for attempt in (1, 2):
|
||||
try:
|
||||
if self.mic is None:
|
||||
self.mic = Mic(self.path)
|
||||
self.info = self.mic.info()
|
||||
return fn(self.mic)
|
||||
except DeviceError:
|
||||
if self.mic is not None:
|
||||
try:
|
||||
self.mic.close()
|
||||
except OSError:
|
||||
pass
|
||||
self.mic = None
|
||||
if attempt == 2:
|
||||
raise
|
||||
|
||||
def snapshot(self):
|
||||
try:
|
||||
state = self.call(lambda m: m.state())
|
||||
return {"connected": True, "info": self.info, "state": state}
|
||||
except DeviceError as e:
|
||||
return {"connected": False, "error": str(e)}
|
||||
|
||||
|
||||
def schema():
|
||||
out = {}
|
||||
for name, (_, enc, desc) in SETTINGS.items():
|
||||
kind = {"type": "range", "min": enc[0], "max": enc[1]} if isinstance(enc, tuple) \
|
||||
else {"type": "toggle"}
|
||||
out[name] = {**kind, "label": desc, "accepts": describe(enc)}
|
||||
return out
|
||||
|
||||
|
||||
def make_handler(holder, allowed_hosts):
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
server_version = "dgm20-web"
|
||||
|
||||
def log_message(self, format, *args): # noqa: A002 (stdlib signature)
|
||||
pass
|
||||
|
||||
def send(self, code, body, ctype="application/json"):
|
||||
if not isinstance(body, bytes):
|
||||
body = json.dumps(body).encode()
|
||||
self.send_response(code)
|
||||
self.send_header("Content-Type", ctype)
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.send_header("X-Content-Type-Options", "nosniff")
|
||||
self.send_header("Content-Security-Policy",
|
||||
"default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; "
|
||||
"script-src 'self' 'unsafe-inline'; frame-ancestors 'none'")
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def host_ok(self):
|
||||
host = (self.headers.get("Host") or "").rsplit(":", 1)[0].strip("[]")
|
||||
if host in allowed_hosts:
|
||||
return True
|
||||
self.send(403, {"error": "unexpected Host header"})
|
||||
return False
|
||||
|
||||
def do_GET(self):
|
||||
if not self.host_ok():
|
||||
return
|
||||
if self.path in ("/", "/index.html"):
|
||||
self.send(200, INDEX, "text/html; charset=utf-8")
|
||||
elif self.path == "/api/state":
|
||||
self.send(200, holder.snapshot())
|
||||
elif self.path == "/api/schema":
|
||||
self.send(200, schema())
|
||||
else:
|
||||
self.send(404, {"error": "not found"})
|
||||
|
||||
def do_POST(self):
|
||||
if not self.host_ok():
|
||||
return
|
||||
if self.headers.get("Content-Type", "").split(";")[0] != "application/json":
|
||||
return self.send(415, {"error": "send application/json"})
|
||||
try:
|
||||
n = int(self.headers.get("Content-Length") or 0)
|
||||
body = json.loads(self.rfile.read(min(n, 4096)) or b"{}")
|
||||
except (ValueError, json.JSONDecodeError):
|
||||
return self.send(400, {"error": "invalid JSON"})
|
||||
try:
|
||||
if self.path == "/api/set":
|
||||
name, value = body.get("name"), body.get("value")
|
||||
if name not in SETTINGS:
|
||||
return self.send(400, {"error": f"unknown setting {name!r}"})
|
||||
holder.call(lambda m: m.set(name, value))
|
||||
self.send(200, holder.snapshot())
|
||||
elif self.path == "/api/factory-reset":
|
||||
if body.get("confirm") is not True:
|
||||
return self.send(400, {"error": "confirm required"})
|
||||
ok = holder.call(lambda m: m.factory_reset())
|
||||
self.send(200 if ok else 502, {"ok": ok})
|
||||
else:
|
||||
self.send(404, {"error": "not found"})
|
||||
except ValueError as e:
|
||||
self.send(400, {"error": str(e)})
|
||||
except DeviceError as e:
|
||||
self.send(503, {"error": str(e)})
|
||||
|
||||
return Handler
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
ap = argparse.ArgumentParser(prog="dgm20-web", description="Local web UI for the Maono DGM20.")
|
||||
ap.add_argument("--host", default="127.0.0.1",
|
||||
help="address to listen on (default 127.0.0.1; use 0.0.0.0 for your LAN, "
|
||||
"there is no login)")
|
||||
ap.add_argument("--port", type=int, default=8720)
|
||||
ap.add_argument("--dev", help="hidraw node (default: autodetect)")
|
||||
ap.add_argument("--open", action="store_true", help="open the page in your browser")
|
||||
a = ap.parse_args(argv)
|
||||
|
||||
allowed = {"localhost", "127.0.0.1", "::1"}
|
||||
if a.host not in ("127.0.0.1", "localhost", "::1"):
|
||||
import socket
|
||||
allowed |= {a.host, socket.gethostname(), socket.getfqdn()}
|
||||
if a.host in ("0.0.0.0", "::"):
|
||||
allowed |= {ip for ip in socket.gethostbyname_ex(socket.gethostname())[2]}
|
||||
allowed |= set(_local_ips())
|
||||
try:
|
||||
srv = ThreadingHTTPServer((a.host, a.port), make_handler(MicHolder(a.dev), allowed))
|
||||
except OSError as e:
|
||||
raise SystemExit(f"dgm20-web: cannot listen on {a.host}:{a.port}: {e.strerror}"
|
||||
" (already running? try --port)") from None
|
||||
url = f"http://{'localhost' if a.host in ('127.0.0.1', '0.0.0.0', '::') else a.host}:{a.port}/"
|
||||
print(f"dgm20-web listening on {url} (ctrl-c to stop)")
|
||||
if a.open:
|
||||
webbrowser.open(url)
|
||||
try:
|
||||
srv.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
|
||||
def _local_ips():
|
||||
import subprocess
|
||||
try:
|
||||
out = subprocess.run(["hostname", "-I"], capture_output=True, text=True, timeout=2).stdout
|
||||
return out.split()
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return []
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user