setup compression using the relay server for back and forth audio communication
This commit is contained in:
Lucca Pirovano
+111
-5
@@ -11,6 +11,12 @@ its whole transfer time, which is a bad tradeoff for something as
|
||||
latency-sensitive as audio. Splitting them means a slow video frame
|
||||
can never delay audio.
|
||||
|
||||
The audio connection is bidirectional: this side keeps writing the
|
||||
robot's own mic (TYPE_AUDIO) to the client, while a reader thread on
|
||||
the same socket listens for TYPE_TALK chunks pushed the other way
|
||||
(push-to-talk audio from the client's mic) and plays them on the
|
||||
robot's speaker via TalkPlayer. Video stays one-way, server->client.
|
||||
|
||||
Why not HTTP: the phone tunnel only forwards specific TCP ports, so
|
||||
this still rides over TCP - but a 5-byte binary header has none of
|
||||
HTTP's chunked/multipart-boundary parsing overhead, and it's trivial
|
||||
@@ -25,12 +31,13 @@ tunnel, half the size of raw PCM.
|
||||
|
||||
Wire format, identical on both connections (all integers big-endian,
|
||||
via `struct`):
|
||||
1 byte type 0x00 HELLO 0x01 VIDEO 0x02 AUDIO
|
||||
1 byte type 0x00 HELLO 0x01 VIDEO 0x02 AUDIO 0x03 TALK
|
||||
4 bytes length
|
||||
N bytes payload
|
||||
HELLO payload: JSON, informational only
|
||||
VIDEO payload: JPEG bytes (video connection only)
|
||||
AUDIO payload: mu-law encoded 8-bit samples, mono (audio connection only)
|
||||
VIDEO payload: JPEG bytes (video connection, server->client only)
|
||||
AUDIO payload: mu-law 8-bit samples, mono (audio connection, server->client - robot's mic)
|
||||
TALK payload: mu-law 8-bit samples, mono (audio connection, client->server - push-to-talk)
|
||||
|
||||
Under load (slow/lossy tunnel), video quality and frame rate step
|
||||
down automatically (see QUALITY_TIERS) and recover once sends are
|
||||
@@ -52,6 +59,7 @@ import collections
|
||||
import json
|
||||
import socket
|
||||
import struct
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
|
||||
@@ -87,6 +95,7 @@ FAST_STREAK_TO_RECOVER = 30 # consecutive quick sends before stepping back
|
||||
TYPE_HELLO = 0x00
|
||||
TYPE_VIDEO = 0x01
|
||||
TYPE_AUDIO = 0x02
|
||||
TYPE_TALK = 0x03
|
||||
|
||||
|
||||
def encode_ulaw(pcm16):
|
||||
@@ -99,6 +108,17 @@ def encode_ulaw(pcm16):
|
||||
return (((y + 1.0) / 2.0) * 255.0).round().astype(np.uint8)
|
||||
|
||||
|
||||
def decode_ulaw(q):
|
||||
"""uint8 mu-law samples -> int16 PCM. Inverse of encode_ulaw()
|
||||
above, used to play back incoming push-to-talk (TYPE_TALK) chunks.
|
||||
Must match naowalk.py's encode_ulaw() exactly, or push-to-talk
|
||||
just produces noise on the robot's speaker."""
|
||||
mu = 255.0
|
||||
y = (q.astype(np.float64) / 255.0) * 2.0 - 1.0
|
||||
x = np.sign(y) * (1.0 / mu) * (np.power(1.0 + mu, np.abs(y)) - 1.0)
|
||||
return np.clip(x * 32768.0, -32768, 32767).astype(np.int16)
|
||||
|
||||
|
||||
class FrameGrabber(object):
|
||||
"""Background thread: keeps the single most recent JPEG-encoded
|
||||
frame available, at whatever quality tier the link currently
|
||||
@@ -218,6 +238,64 @@ class MicGrabber(object):
|
||||
pass
|
||||
|
||||
|
||||
class TalkPlayer(object):
|
||||
"""Decodes incoming TYPE_TALK chunks (push-to-talk audio pushed
|
||||
from the client) and plays them on the robot's speaker.
|
||||
|
||||
Uses a persistent `aplay` subprocess fed raw PCM over its stdin -
|
||||
the same playback mechanism the old SSH `nc -l | aplay` relay
|
||||
used, just sourced from decoded network chunks instead of a raw
|
||||
pipe. The process is kept open across calls (silence between
|
||||
presses is fine); it's only recreated if it exited or hasn't
|
||||
been started yet."""
|
||||
|
||||
def __init__(self, rate=AUDIO_RATE):
|
||||
self.rate = rate
|
||||
self._proc = None
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def _ensure_proc(self):
|
||||
if self._proc is None or self._proc.poll() is not None:
|
||||
self._proc = subprocess.Popen(
|
||||
["aplay", "-q", "-t", "raw", "-f", "S16_LE",
|
||||
"-c", str(AUDIO_CHANNELS), "-r", str(self.rate)],
|
||||
stdin=subprocess.PIPE)
|
||||
|
||||
def play(self, ulaw_payload):
|
||||
pcm = decode_ulaw(np.frombuffer(ulaw_payload, dtype=np.uint8))
|
||||
with self._lock:
|
||||
try:
|
||||
self._ensure_proc()
|
||||
self._proc.stdin.write(pcm.tobytes())
|
||||
self._proc.stdin.flush()
|
||||
except Exception as e:
|
||||
print("talk playback error: %s" % e)
|
||||
self._proc = None
|
||||
|
||||
def close(self):
|
||||
with self._lock:
|
||||
if self._proc is not None:
|
||||
try:
|
||||
self._proc.stdin.close()
|
||||
self._proc.terminate()
|
||||
except Exception:
|
||||
pass
|
||||
self._proc = None
|
||||
|
||||
|
||||
def _recv_exact(sock, n):
|
||||
"""Read exactly n bytes from a TCP socket (recv() can return short
|
||||
reads/chunks at any point - never assume one call gives you n).
|
||||
Mirror of naowalk.py's _recv_exact()."""
|
||||
buf = b""
|
||||
while len(buf) < n:
|
||||
chunk = sock.recv(n - len(buf))
|
||||
if not chunk:
|
||||
raise IOError("media relay connection closed")
|
||||
buf += chunk
|
||||
return buf
|
||||
|
||||
|
||||
def _send(conn, msg_type, payload):
|
||||
conn.sendall(struct.pack(">BI", msg_type, len(payload)))
|
||||
conn.sendall(payload)
|
||||
@@ -263,7 +341,29 @@ def serve_video(grabber, host, port):
|
||||
pass
|
||||
|
||||
|
||||
def serve_audio(mic, host, port):
|
||||
def _talk_reader(conn, player):
|
||||
"""Runs alongside the audio writer loop on the same connection -
|
||||
the audio socket is bidirectional now, so this thread's the read
|
||||
half: it pulls TYPE_TALK frames pushed from the client (push-to-
|
||||
talk) and hands each decoded chunk to TalkPlayer. conn has a
|
||||
socket timeout set by the caller, so a quiet stretch between
|
||||
button presses just means occasional socket.timeout - not a
|
||||
disconnect - and the loop keeps going."""
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
header = _recv_exact(conn, 5)
|
||||
except socket.timeout:
|
||||
continue
|
||||
msg_type, length = struct.unpack(">BI", header)
|
||||
payload = _recv_exact(conn, length)
|
||||
if msg_type == TYPE_TALK:
|
||||
player.play(payload)
|
||||
except Exception as e:
|
||||
print("talk reader stopped (%s)" % e)
|
||||
|
||||
|
||||
def serve_audio(mic, player, host, port):
|
||||
"""Audio's own connection/thread. Checked eagerly (short sleep,
|
||||
not paced like video) since gaps are far more noticeable than a
|
||||
video frame being a beat stale."""
|
||||
@@ -277,6 +377,9 @@ def serve_audio(mic, host, port):
|
||||
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
||||
conn.settimeout(5.0)
|
||||
print("audio client connected: %s" % (addr,))
|
||||
reader_thread = threading.Thread(target=_talk_reader, args=(conn, player))
|
||||
reader_thread.daemon = True
|
||||
reader_thread.start()
|
||||
try:
|
||||
_send(conn, TYPE_HELLO,
|
||||
json.dumps({"rate": AUDIO_RATE, "channels": AUDIO_CHANNELS}).encode("utf-8"))
|
||||
@@ -313,14 +416,17 @@ def main():
|
||||
video_thread.daemon = True
|
||||
video_thread.start()
|
||||
|
||||
player = TalkPlayer()
|
||||
|
||||
try:
|
||||
serve_audio(mic, "0.0.0.0", AUDIO_PORT)
|
||||
serve_audio(mic, player, "0.0.0.0", AUDIO_PORT)
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
grabber.close()
|
||||
if mic:
|
||||
mic.close()
|
||||
player.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user