commit d239c214b48889e1b621422990942bf5fa3a2d67 Author: Justin Reiners Date: Wed Oct 7 22:45:59 2026 -0500 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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..47584ad --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.egg-info/ +build/ +dist/ +.venv/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..a978c00 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Justin Reiners + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/PROTOCOL.md b/PROTOCOL.md new file mode 100644 index 0000000..587bf0b --- /dev/null +++ b/PROTOCOL.md @@ -0,0 +1,74 @@ +# DGM20 control protocol + +These notes come from Maono Link 4.0.88 for Windows. The app loads a per-model plugin, and the +plugin for USB `352f:0105` is `DGM20_OLD.dll`. Every value below was checked against a real mic +(firmware 2.1.2) unless it is marked otherwise. + +## Transport + +The mic has USB audio interfaces 0-2 and a HID interface 3. Interface 3 carries a vendor usage page +(`0xFF01`) with four reports: + +| report | direction | size | +|--------|-----------|------| +| 0x50 | out | 15 | +| 0x51 | out | 63 | +| 0x10 | in | 15 | +| 0x11 | in | 63 | + +The app uses only 0x51 (requests) and 0x11 (replies). On Linux the interface is a hidraw node: a +65-byte `write()` (the report id plus 64 data bytes) sends a request, and `read()` returns replies. + +## Packet + +``` +offset size field +0 1 report id: 0x51 request, 0x11 reply +1 1 0xC4 +2 1 length of bytes 1..end, including the checksum +3 2 0x0000 +5 1 type +6 n payload, u16 little-endian words +6+n 2 checksum, u16 LE = -(sum of bytes 1 .. 5+n) +``` + +The rest of the 64 bytes is zero-padded. + +| type | request payload | reply | +|------|-----------------|-------| +| 2 | (reply only) | write ack: `status` (0 = ok) | +| 3 | `addr, value` pairs | type-2 ack. The mic also sends type-3 packets on its own when a setting changes on the hardware, such as the touch pad. | +| 4 | `addr` | `addr, value` | +| 5 | `addr, count, data...` | bulk write. The app uses it only to write the serial number; don't. | +| 6 | `addr, count` | `addr, count, values...` (range read) | + +## Registers + +| reg | meaning | values | +|--------|---------|--------| +| 0x0003 | model name, 8 words, ASCII | `"DGM20"` | +| 0x000B | firmware version, 3 words | `2, 1, 2` | +| 0x001E | serial number, up to 20 words | | +| 0x1008 | DFU mode (read) | 0 = normal | +| 0x100A | factory reset | write 1 | +| 0x2022 | mute | 0 / 1 | +| 0x2026 | headphone monitoring | 2000 = on, 3000 = off | +| 0x2027 | noise cancellation | 0 / 1 | +| 0x2028 | noise cancellation strength | 0-100 | +| 0x202A | RGB ring brightness | 0-100 (the app does not show it for this model) | +| 0x202C | vocal enhancement | 0 / 1 | +| 0x204B | level-meter enable | not answered by firmware 2.1.2 | + +Maono Link reads settings once, in this order: 0x204B, then 0x2022, then a range read of 0x2027 +(count 2), then 0x2026, then 0x202C. After that it relies on type-3 packets the mic sends on its +own. + +## Present in the plugin but not used for this model + +The DLL shares widgets with the PD200X plugin, so it contains EQ and RGB-colour code. That code +is never instantiated for the DGM20, and the mic does not answer reads of these registers. They +are listed here for anyone exploring. + +- **EQ:** 5 parametric bands, each 5 words, starting at 0x204D + 5·i. The HPF is at 0x2066 and the + LPF at 0x206B. Each band is `[enable, type, freq Hz, gain dB×10+2000, Q×100]`. +- **Light on/off:** 0x2036. **Colour index:** 0x2038. diff --git a/README.md b/README.md new file mode 100644 index 0000000..431195b --- /dev/null +++ b/README.md @@ -0,0 +1,78 @@ +# dgm20-linux + +Control the **Maono DGM20** USB microphone on Linux. Maono Link only runs on Windows and macOS; this +covers the same settings with a local web page and a command-line tool. + +![The control page](docs/screenshot.png) + +- Noise cancellation on/off and strength +- Vocal enhancement +- Hear yourself (headphone monitoring) +- Mute, kept in sync with the mic's touch pad +- RGB ring brightness +- Factory reset + +Pure Python 3.9+, no dependencies. Nothing leaves your machine: the page is served from localhost +and loads no external fonts or scripts. + +> Unofficial. Not affiliated with or endorsed by Maono. Tested on a DGM20 with USB ID +> `352f:0105`, firmware 2.1.2. Newer DGM20 revisions (`352f:0112`) and the DGM20G/DGM20W +> use different firmware and are not supported yet. + +## Install + +```sh +pipx install git+https://git.reiners.dev/ReinersCloudConsulting/dgm20-linux.git +# or, from a checkout +pipx install . +``` + +Let your user reach the mic without root, then unplug and replug it: + +```sh +sudo cp udev/70-maono-dgm20.rules /etc/udev/rules.d/ +sudo udevadm control --reload && sudo udevadm trigger +``` + +The rule grants access to the logged-in desktop user (`uaccess`) and to the `wheel` group, which +covers ssh sessions. Change the group if your distro uses a different one. + +## Use + +```sh +dgm20-web --open # control page at http://localhost:8720 +dgm20ctl # show settings +dgm20ctl set nc on nc-level 80 enhance off +dgm20ctl watch # print changes made on the mic itself +dgm20ctl factory-reset --yes +``` + +To keep the page running in the background: + +```sh +cp contrib/dgm20-web.service ~/.config/systemd/user/ +systemctl --user enable --now dgm20-web +``` + +`dgm20-web` only listens on localhost. `--host 0.0.0.0` makes it reachable from your network, and +anyone on that network can then change your mic settings, because there is no login. + +Input gain and headphone volume are standard USB audio controls. Set them in your desktop's sound +settings, `pavucontrol` or `alsamixer`. The mic does not expose them on its control channel. + +## How it works + +The settings were found by reading Maono Link's DGM20 plugin. See [PROTOCOL.md](PROTOCOL.md) for +the packet format and register map. `dgm20ctl raw` gives direct register access if you want to +explore. + +## Development + +```sh +PYTHONPATH=src python3 -m unittest discover -s tests +PYTHONPATH=src python3 -m dgm20.web +``` + +## License + +MIT diff --git a/contrib/dgm20-web.service b/contrib/dgm20-web.service new file mode 100644 index 0000000..9aca271 --- /dev/null +++ b/contrib/dgm20-web.service @@ -0,0 +1,9 @@ +[Unit] +Description=Maono DGM20 web control panel (http://localhost:8720) + +[Service] +ExecStart=%h/.local/bin/dgm20-web +Restart=on-failure + +[Install] +WantedBy=default.target diff --git a/docs/screenshot.png b/docs/screenshot.png new file mode 100644 index 0000000..458eed0 Binary files /dev/null and b/docs/screenshot.png differ diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9dd91ec --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools>=64"] +build-backend = "setuptools.build_meta" + +[project] +name = "dgm20-linux" +version = "0.1.0" +description = "Control the Maono DGM20 USB microphone on Linux: noise cancellation, monitoring, vocal enhancement, mute, RGB brightness" +readme = "README.md" +license = "MIT" +requires-python = ">=3.9" +dependencies = [] +classifiers = ["Operating System :: POSIX :: Linux", "Topic :: Multimedia :: Sound/Audio"] + +[project.scripts] +dgm20ctl = "dgm20.cli:main" +dgm20-web = "dgm20.web:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +dgm20 = ["static/*.html"] diff --git a/src/dgm20/__init__.py b/src/dgm20/__init__.py new file mode 100644 index 0000000..6919f31 --- /dev/null +++ b/src/dgm20/__init__.py @@ -0,0 +1 @@ +"""Linux control for the Maono DGM20 USB microphone.""" diff --git a/src/dgm20/cli.py b/src/dgm20/cli.py new file mode 100644 index 0000000..7cdb30f --- /dev/null +++ b/src/dgm20/cli.py @@ -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() diff --git a/src/dgm20/device.py b/src/dgm20/device.py new file mode 100644 index 0000000..e2e7c17 --- /dev/null +++ b/src/dgm20/device.py @@ -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) diff --git a/src/dgm20/static/index.html b/src/dgm20/static/index.html new file mode 100644 index 0000000..a007c6b --- /dev/null +++ b/src/dgm20/static/index.html @@ -0,0 +1,337 @@ + + + + + +DGM20 Control + + + + +
+
+

DGM20 control

+
Connecting
+
+ + + +
+ +

Waiting for mic

+

 

+
+ +
    +
  • +

    Noise cancellation

    +

    Filters out keyboard clatter, fans and room hum.

    + +
    • +
  • + + + – +
    • +
  • +

    Vocal enhancement

    +

    Brings your voice forward over background sound.

    + +
    • +
  • +

    Hear yourself

    +

    Plays the mic through headphones plugged into the DGM20.

    + +
    • +
  • + + + – +
    • +
+ +
+ Unofficial tool, not affiliated with Maono. + + + + +
+
+ + + + diff --git a/src/dgm20/web.py b/src/dgm20/web.py new file mode 100644 index 0000000..7977f0f --- /dev/null +++ b/src/dgm20/web.py @@ -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() diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..3a19c23 --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,64 @@ +"""Framing tests against packets captured from a real DGM20 (firmware 2.1.2).""" +import unittest + +from dgm20.device import decode, encode, frame, parse, words_to_ascii + + +def h(s): + return bytes.fromhex(s) + bytes(64 - len(bytes.fromhex(s))) + + +class Framing(unittest.TestCase): + def test_read_request_matches_app(self): + # checksum = -(0xC4+0x09+0x04+0x28+0x20) = -0x119 = 0xFEE7 + p = frame(4, [0x2028]) + self.assertEqual(p[:10].hex(" "), "51 c4 09 00 00 04 28 20 e7 fe") + self.assertEqual(len(p), 65) + + def test_write_request_checksum(self): + p = frame(3, [0x2028, 64]) + self.assertEqual(p[1:3], bytes([0xC4, 0x0B])) + self.assertEqual((sum(p[1:10]) + (p[10] | p[11] << 8)) & 0xFFFF, 0) + + def test_parse_read_reply(self): + self.assertEqual(parse(h("11 c4 0b 00 00 04 28 20 40 00 a5 fe")), (4, [0x2028, 64])) + + def test_parse_range_reply(self): + self.assertEqual(parse(h("11 c4 0f 00 00 06 27 20 02 00 01 00 40 00 9d fe")), + (6, [0x2027, 2, 1, 64])) + + def test_parse_ack(self): + self.assertEqual(parse(h("11 c4 09 00 00 02 00 00 31 ff")), (2, [0])) + + def test_parse_rejects_bad_checksum(self): + self.assertIsNone(parse(h("11 c4 0b 00 00 04 28 20 40 00 a5 ff"))) + + def test_model_name(self): + self.assertEqual(words_to_ascii([0x4744, 0x324D, 0x0030, 0]), "DGM20") + + +class Encoding(unittest.TestCase): + def test_bool(self): + self.assertEqual(encode("nc", "on"), 1) + self.assertEqual(encode("mute", False), 0) + self.assertIs(decode("nc", 1), True) + + def test_monitor_uses_2000_3000(self): + self.assertEqual(encode("monitor", True), 2000) + self.assertEqual(encode("monitor", "off"), 3000) + self.assertEqual(decode("monitor", 2000), "on") + + def test_range(self): + self.assertEqual(encode("nc-level", "0"), 0) + self.assertEqual(encode("nc-level", 100), 100) + for bad in (101, -1, True, "loud"): + with self.assertRaises(ValueError): + encode("nc-level", bad) + + def test_unknown(self): + with self.assertRaises(ValueError): + encode("volume", 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/udev/70-maono-dgm20.rules b/udev/70-maono-dgm20.rules new file mode 100644 index 0000000..0f90c7c --- /dev/null +++ b/udev/70-maono-dgm20.rules @@ -0,0 +1,3 @@ +# Maono DGM20 USB microphone: vendor HID channel for dgm20ctl. +# uaccess covers local desktop logins; the group covers ssh/tmux sessions. +SUBSYSTEM=="hidraw", ATTRS{idVendor}=="352f", ATTRS{idProduct}=="0105", MODE="0660", GROUP="wheel", TAG+="uaccess"