mirror of
https://github.com/MCKero6423/uv-k5-v3-emulator.git
synced 2026-10-02 11:07:31 +00:00
Model the audio path, which is a single enable line and no samples
Asked for a speaker and a microphone. The honest answer is that neither exists to
model: on the real radio neither passes through the MCU. Receive audio is demodulated
inside the BK4819 and leaves it as analogue on its AF pin; transmit audio goes from
the microphone into the chip's own ADC. The firmware's entire involvement is
- PA8, the amplifier enable (GPIO_EnableAudioPath, driver/gpio.h:34)
- REG_47, which AF source the chip routes
- REG_64, a level it displays
No audio samples exist anywhere in the MCU's address space, so a device model has
nothing to capture or play, and a browser has nothing to be granted permission for.
Synthesising sound would be inventing data the firmware never produced.
What is real is the firmware's intent, and PA8 states it exactly. TYPE_UVK5_AUDIO
watches that pin and exposes read-only speaker-on; the web UI shows it as a speaker
glyph beside the power state, and /api/status reports it. Read-only deliberately:
letting a test write it would only let the test lie to itself. The page asks for no
audio permission, and a test asserts it never will -- no getUserMedia, no
AudioContext, no <audio>.
Measured: amplifier off while idle in power save, on after SIDE1 engages monitor, and
still on afterwards rather than blipping.
One bug found the hard way, and the stub was the cause. QmpClient.command returns the
unwrapped value and raises on error, but the test stub returned {"return": ...}. So
webui.py was written to unwrap a second time, every test passed, and the live UI
returned 500 with "argument of type 'bool' is not iterable". The stub is now pinned to
the real contract by a test. A stub more forgiving than the real thing is worse than
no stub.
Also stopped swallowing the failure: a bare `except: return None` made the error
indistinguishable from a radio that simply was not making sound, and cost a detour
into looking for a stale process.
This commit is contained in:
1 parent
8d1a1c4415
commit
7d299dbad2
4 files changed
+355
-1
No files matched your search
Executable
+153
@@ -0,0 +1,153 @@
|
||||
#!/usr/bin/env python3
|
||||
"""The audio amplifier must turn on when the firmware decides to make sound.
|
||||
|
||||
Scope, stated plainly, because "add a speaker and a microphone" is the obvious request
|
||||
and the honest answer is that there is nothing to add. On the real radio neither passes
|
||||
through the MCU:
|
||||
|
||||
* receive audio is demodulated inside the BK4819 and leaves it as analogue on its AF
|
||||
output pin, going straight to the amplifier
|
||||
* transmit audio goes from the microphone into the chip's own ADC
|
||||
|
||||
The firmware's whole involvement is PA8 (amplifier enable), REG_47 (which AF source the
|
||||
chip routes) and REG_64 (a level it displays). No audio samples exist anywhere in the
|
||||
MCU's address space, so a device model has nothing to capture or play, and browser audio
|
||||
permissions have nothing to carry. Synthesising sound would be inventing data the
|
||||
firmware never produced.
|
||||
|
||||
What is real is the firmware's *intent*, and PA8 states it exactly. Checked here:
|
||||
|
||||
1. the amplifier is off while idle in power save
|
||||
2. engaging monitor (SIDE1) turns it on -- squelch forced open means listening
|
||||
3. it is still on afterwards, i.e. this is a state and not a blip
|
||||
"""
|
||||
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
SIM = pathlib.Path(__file__).resolve().parent.parent
|
||||
QEMU = pathlib.Path(os.environ.get(
|
||||
"QEMU", "/root/qemu-build/qemu-7.2+dfsg/build/qemu-system-arm"))
|
||||
ELF = pathlib.Path(os.environ.get(
|
||||
"ELF", "/root/uvk5-port/uvk5-sat/build/CW/nr7y.cw.elf"))
|
||||
PRISTINE = SIM / "assets/pristine/flash-pristine.img.gz"
|
||||
|
||||
BOOT_SECONDS = 24
|
||||
|
||||
|
||||
class Qmp:
|
||||
def __init__(self, path):
|
||||
self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
self.s.settimeout(25)
|
||||
self.s.connect(path)
|
||||
self.buf = b""
|
||||
self._read()
|
||||
self.cmd("qmp_capabilities")
|
||||
|
||||
def _read(self):
|
||||
while b"\n" not in self.buf:
|
||||
chunk = self.s.recv(65536)
|
||||
if not chunk:
|
||||
raise RuntimeError("QMP closed")
|
||||
self.buf += chunk
|
||||
line, self.buf = self.buf.split(b"\n", 1)
|
||||
return json.loads(line)
|
||||
|
||||
def cmd(self, name, **args):
|
||||
msg = {"execute": name}
|
||||
if args:
|
||||
msg["arguments"] = args
|
||||
self.s.sendall(json.dumps(msg).encode() + b"\n")
|
||||
while True:
|
||||
reply = self._read()
|
||||
if "return" in reply or "error" in reply:
|
||||
return reply
|
||||
|
||||
def speaker_on(self):
|
||||
return self.cmd("qom-get", path="/machine/audio",
|
||||
property="speaker-on").get("return")
|
||||
|
||||
def key(self, name, hold=0.15):
|
||||
self.cmd("qom-set", path="/machine/keypad", property="press", value=name)
|
||||
time.sleep(hold)
|
||||
self.cmd("qom-set", path="/machine/keypad", property="press", value="")
|
||||
|
||||
|
||||
def main():
|
||||
for tool in (QEMU, ELF, PRISTINE):
|
||||
if not tool.exists():
|
||||
print(f"SKIP missing {tool}")
|
||||
return 0
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
img = pathlib.Path(tmp) / "flash.img"
|
||||
img.write_bytes(gzip.decompress(PRISTINE.read_bytes()))
|
||||
sock = pathlib.Path(tmp) / "qmp.sock"
|
||||
|
||||
proc = subprocess.Popen(
|
||||
[str(QEMU), "-M", f"uv-k5-v3,flash-image={img}",
|
||||
"-nographic", "-monitor", "none",
|
||||
"-qmp", f"unix:{sock},server=on,wait=off",
|
||||
"-kernel", str(ELF)],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
try:
|
||||
for _ in range(BOOT_SECONDS * 4):
|
||||
if sock.exists():
|
||||
break
|
||||
time.sleep(0.25)
|
||||
else:
|
||||
print("FAIL QMP socket never appeared")
|
||||
return 1
|
||||
time.sleep(BOOT_SECONDS)
|
||||
|
||||
qmp = Qmp(str(sock))
|
||||
failures = 0
|
||||
|
||||
idle = qmp.speaker_on()
|
||||
print(f"idle: speaker-on={idle}")
|
||||
if idle:
|
||||
print("FAIL the amplifier is on while the radio sits in power save")
|
||||
failures += 1
|
||||
else:
|
||||
print("PASS the amplifier is off when there is nothing to hear")
|
||||
|
||||
qmp.key("SIDE1")
|
||||
time.sleep(3)
|
||||
listening = qmp.speaker_on()
|
||||
print(f"monitoring: speaker-on={listening}")
|
||||
if listening:
|
||||
print("PASS engaging monitor turned the amplifier on")
|
||||
else:
|
||||
print("FAIL monitor is engaged but the amplifier stayed off")
|
||||
failures += 1
|
||||
|
||||
time.sleep(2)
|
||||
still = qmp.speaker_on()
|
||||
print(f"still listening: speaker-on={still}")
|
||||
if still:
|
||||
print("PASS it stays on; this is a state, not a blip")
|
||||
else:
|
||||
print("FAIL the amplifier dropped again straight away")
|
||||
failures += 1
|
||||
|
||||
if failures:
|
||||
return 1
|
||||
print("\nthe firmware's audio intent is observable")
|
||||
return 0
|
||||
finally:
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -11,11 +11,20 @@ class StubClient:
|
||||
|
||||
def __init__(self):
|
||||
self.sent = []
|
||||
self.speaker = False
|
||||
|
||||
def command(self, name, **args):
|
||||
self.sent.append((name, args))
|
||||
if name == "query-status":
|
||||
return {"status": "running", "running": True}
|
||||
if name == "qom-get" and args.get("property") == "speaker-on":
|
||||
# The real client unwraps: QmpClient.command returns msg["return"]
|
||||
# directly and raises on error. An earlier version of this stub returned
|
||||
# {"return": ...}, so webui.py was written to unwrap a second time and blew
|
||||
# up with "argument of type 'bool' is not iterable" against a live guest,
|
||||
# while the tests passed. A stub that is more forgiving than the real thing
|
||||
# is worse than no stub.
|
||||
return self.speaker
|
||||
if name == "memsave":
|
||||
with open(args["filename"], "wb") as fh:
|
||||
fh.write(bytes(args["size"]))
|
||||
@@ -733,3 +742,52 @@ class TestClientIpInLogs(unittest.TestCase):
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
class TestSpeakerIndicator(unittest.TestCase):
|
||||
"""The UI reports whether the firmware wants sound.
|
||||
|
||||
Deliberately not audio. The microphone and speaker are wired to the BK4819 rather
|
||||
than the MCU, so no samples reach the emulator, there is nothing to stream, and the
|
||||
page needs no audio permission. What is reported is the amplifier enable (PA8), i.e.
|
||||
the firmware's intent.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.client, self.http = make_app()
|
||||
|
||||
def test_stub_matches_the_real_client_contract(self):
|
||||
"""qom-get must yield the value itself, not a {"return": ...} envelope.
|
||||
|
||||
This pins the stub to QmpClient.command's actual behaviour. When they diverged,
|
||||
every test passed and the live web UI returned 500.
|
||||
"""
|
||||
result = self.client.command("qom-get", path="/machine/audio",
|
||||
property="speaker-on")
|
||||
self.assertIsInstance(result, bool)
|
||||
|
||||
def test_status_reports_the_speaker_state(self):
|
||||
self.client.speaker = True
|
||||
body = self.http.get("/api/status").get_json()
|
||||
self.assertTrue(body["speaker"])
|
||||
|
||||
def test_status_reports_a_silent_radio(self):
|
||||
self.client.speaker = False
|
||||
body = self.http.get("/api/status").get_json()
|
||||
self.assertFalse(body["speaker"])
|
||||
|
||||
def test_page_has_an_indicator_wired_to_the_poll(self):
|
||||
body = self.http.get("/").get_data(as_text=True)
|
||||
self.assertIn('id="speaker"', body)
|
||||
self.assertIn("showSpeaker", body)
|
||||
|
||||
def test_page_requests_no_audio_permission(self):
|
||||
"""Nothing here should ask the browser for a microphone or to play audio.
|
||||
|
||||
There is no audio to carry, so a permission prompt would be asking the user to
|
||||
approve something that cannot happen.
|
||||
"""
|
||||
body = self.http.get("/").get_data(as_text=True)
|
||||
for forbidden in ("getUserMedia", "AudioContext", "navigator.mediaDevices",
|
||||
"new Audio", "<audio"):
|
||||
self.assertNotIn(forbidden, body)
|
||||
+45
-1
@@ -29,6 +29,7 @@ from uvk5_logs import LogBuffer
|
||||
from uvk5_stream import FramePump
|
||||
|
||||
KEYPAD_PATH = "/machine/keypad"
|
||||
AUDIO_PATH = "/machine/audio"
|
||||
|
||||
# Firmware thresholds, from App/misc.c:
|
||||
# key_debounce_10ms = 2 -> 20 ms to register a press
|
||||
@@ -154,6 +155,30 @@ def create_app(client, frame_addr: int, status_addr: int, scale: int = 4,
|
||||
raise LookupError("emulator is off")
|
||||
target.command("qom-set", path=KEYPAD_PATH, property="press", value=value)
|
||||
|
||||
def speaker_on():
|
||||
"""Whether the firmware has enabled the audio amplifier (PA8).
|
||||
|
||||
Not audio. The microphone and speaker are wired to the BK4819, not to the MCU,
|
||||
so no samples pass through the emulator and there is nothing to stream to a
|
||||
browser -- which is also why this needs no audio permission. What it reports is
|
||||
the firmware's intent: whether the radio would be making sound right now.
|
||||
"""
|
||||
target = active_client()
|
||||
if target is None:
|
||||
return None
|
||||
try:
|
||||
# command() returns the unwrapped value and raises on a QMP error, so there
|
||||
# is no envelope to inspect here. Treating the result as {"return": ...}
|
||||
# raised TypeError: argument of type 'bool' is not iterable.
|
||||
return target.command("qom-get", path=AUDIO_PATH,
|
||||
property="speaker-on")
|
||||
except Exception as exc:
|
||||
# Log rather than swallow. A silent None is indistinguishable from a radio
|
||||
# that simply is not making sound, which sent me looking in the wrong place
|
||||
# once already.
|
||||
log.add("qemu", f"speaker state unavailable: {exc}")
|
||||
return None
|
||||
|
||||
def set_ptt(held: bool):
|
||||
"""Hold or release PTT.
|
||||
|
||||
@@ -180,7 +205,7 @@ def create_app(client, frame_addr: int, status_addr: int, scale: int = 4,
|
||||
except Exception as exc:
|
||||
# The emulator can die under us; that is a state to report, not a 500.
|
||||
return jsonify(powered=False, status="unreachable", error=str(exc))
|
||||
return jsonify(powered=True, **info)
|
||||
return jsonify(powered=True, speaker=speaker_on(), **info)
|
||||
|
||||
@app.get("/api/logs")
|
||||
def api_logs():
|
||||
@@ -424,6 +449,15 @@ def render_index(scale: int) -> str:
|
||||
.pwr:disabled {{ opacity:0.5; cursor:default; }}
|
||||
#powerstate {{ font-size:12px; color:#6e7681; margin-left:auto; }}
|
||||
#powerstate.on {{ color:#3fb950; }}
|
||||
/*
|
||||
* Speaker indicator. Not audio: the microphone and speaker are wired to the BK4819
|
||||
* rather than the MCU, so no samples reach the emulator and there is nothing to play
|
||||
* -- which is why this page asks for no audio permission. It shows whether the
|
||||
* firmware currently has the amplifier enabled, i.e. whether a real radio would be
|
||||
* making sound.
|
||||
*/
|
||||
#speaker {{ font-size:14px; opacity:0.25; transition:opacity 0.15s; }}
|
||||
#speaker.on {{ opacity:1; }}
|
||||
/*
|
||||
* Powered off is a dark panel, drawn by the wrapper so the frame itself can be
|
||||
* hidden. An earlier attempt put a dark background on the <img> alone, which
|
||||
@@ -454,6 +488,7 @@ def render_index(scale: int) -> str:
|
||||
<button class="pwr" data-power="off">Off</button>
|
||||
<button class="pwr" data-power="reset">Reset</button>
|
||||
<span id="powerstate">-</span>
|
||||
<span id="speaker" title="the firmware has enabled the audio amplifier">🔈</span>
|
||||
</div>
|
||||
<div class="screenwrap" id="screenwrap">
|
||||
<img id="screen" src="/stream" alt="radio LCD"
|
||||
@@ -604,6 +639,10 @@ document.querySelectorAll('.pwr').forEach(btn => {{
|
||||
}});
|
||||
}});
|
||||
|
||||
function showSpeaker(on) {{
|
||||
document.getElementById('speaker').classList.toggle('on', !!on);
|
||||
}}
|
||||
|
||||
function showPower(powered) {{
|
||||
const label = document.getElementById('powerstate');
|
||||
label.textContent = powered ? 'on' : 'off';
|
||||
@@ -617,15 +656,20 @@ async function poll() {{
|
||||
const r = await fetch('/api/status');
|
||||
const s = await r.json();
|
||||
showPower(!!s.powered);
|
||||
showSpeaker(s.speaker);
|
||||
document.getElementById('status').textContent =
|
||||
s.powered ? ('guest: ' + (s.status || 'unknown'))
|
||||
: 'powered off -- press On to boot';
|
||||
}} catch (err) {{
|
||||
showPower(false);
|
||||
showSpeaker(false);
|
||||
document.getElementById('status').textContent = 'server unreachable';
|
||||
}}
|
||||
}}
|
||||
poll();
|
||||
// 3 s: the speaker indicator rides this existing request rather than adding another.
|
||||
// It lags a real amplifier transition by up to that long, which is fine for showing
|
||||
// state and would not be for anything timing-sensitive.
|
||||
setInterval(poll, 3000);
|
||||
|
||||
// Cap the DOM as well as the server-side buffer. The pane scrolls, but an
|
||||
|
||||
Reference in new issue
Block a user