Hi ! I’ve been experimenting with a WiFi module to implement data transmission over HTTP. Below is the code I wrote for session-based data transfer.
My receiving end can achieve a data reception speed of 1 MB/s, yet the transmission throughput from my OpenMV N6 barely hits 50 KB/s. I’m reaching out to ask: are there any optimizations or adjustments I can make to the code attached below?
"""OpenMV N6 HTTP client for Meeting ASR 0.5 device sessions.
This module intentionally uses plain sockets and HTTP. Deploy it only on the
trusted LAN configured on the Meeting ASR receiver.
"""
try:
import ujson as json
except ImportError: # CPython tests
import json
import socket
import struct
import time
AV_AUDIO = 1
AV_JPEG = 2
AV_END = 255
AV_PROTOCOL = "mav1"
UPLOAD_CHUNK_BYTES = 1024 * 1024
def _safe_send(sock, data):
"""Reliably send a bytes-like object without making a second copy."""
view = data if isinstance(data, memoryview) else memoryview(data)
sock.sendall(view)
def ticks_ms():
function = getattr(time, "ticks_ms", None)
if function:
return function()
return int(time.monotonic() * 1000)
def ticks_diff(current, previous):
function = getattr(time, "ticks_diff", None)
if function:
return function(current, previous)
return current - previous
def sleep_ms(milliseconds):
function = getattr(time, "sleep_ms", None)
if function:
function(milliseconds)
else:
time.sleep(milliseconds / 1000.0)
def wav_header(pcm_bytes, sample_rate=16000, channels=1, bits_per_sample=16):
"""Return a canonical 44-byte little-endian PCM WAV header."""
byte_rate = sample_rate * channels * bits_per_sample // 8
block_align = channels * bits_per_sample // 8
return struct.pack(
"<4sI4s4sIHHIIHH4sI",
b"RIFF",
36 + pcm_bytes,
b"WAVE",
b"fmt ",
16,
1,
channels,
sample_rate,
byte_rate,
block_align,
bits_per_sample,
b"data",
pcm_bytes,
)
def av_record(record_type, sequence, timestamp_ms, payload=b""):
"""Encode one mav1 record using the fixed !BBHIII header."""
if record_type not in (AV_AUDIO, AV_JPEG, AV_END):
raise ValueError("invalid A/V record type")
if sequence < 0 or sequence > 0xFFFFFFFF:
raise ValueError("invalid A/V sequence")
if timestamp_ms < 0 or timestamp_ms > 0xFFFFFFFF:
raise ValueError("invalid A/V timestamp")
if len(payload) > 512 * 1024:
raise ValueError("A/V payload is too large")
return struct.pack(
">BBHIII",
record_type,
0,
0,
sequence,
timestamp_ms,
len(payload),
) + payload
class PcmRing:
"""Fixed-storage PCM queue safe to fill from the audio callback."""
def __init__(self, capacity=16, slot_bytes=2048):
self.capacity = capacity
self.slot_bytes = slot_bytes
self.slots = [bytearray(slot_bytes) for _ in range(capacity)]
self.lengths = [0] * capacity
self.timestamps = [0] * capacity
self.head = 0
self.tail = 0
self.count = 0
self.overflowed = False
def push(self, payload, timestamp_ms=0):
length = len(payload)
if self.count >= self.capacity or length > self.slot_bytes:
self.overflowed = True
return False
self.slots[self.tail][:length] = payload
self.lengths[self.tail] = length
self.timestamps[self.tail] = timestamp_ms
self.tail = (self.tail + 1) % self.capacity
self.count += 1
return True
def pop(self):
item = self.pop_with_timestamp()
return None if item is None else item[1]
def pop_with_timestamp(self):
if self.count == 0:
return None
index = self.head
length = self.lengths[index]
timestamp_ms = self.timestamps[index]
self.head = (self.head + 1) % self.capacity
self.count -= 1
return timestamp_ms, memoryview(self.slots[index])[:length]
def _parse_http_url(value):
if not value.startswith("http://"):
raise ValueError("Only trusted-LAN http:// server URLs are supported")
authority = value[7:].rstrip("/")
if "/" in authority or "@" in authority:
raise ValueError("SERVER_URL must be an HTTP origin without credentials")
if authority.startswith("["):
closing = authority.find("]")
if closing < 0:
raise ValueError("invalid IPv6 SERVER_URL")
host = authority[1:closing]
port = int(authority[closing + 2 :]) if authority[closing + 1 :] else 80
elif ":" in authority:
host, raw_port = authority.rsplit(":", 1)
port = int(raw_port)
else:
host = authority
port = 80
if not host or port < 1 or port > 65535:
raise ValueError("invalid SERVER_URL")
return host, port
class HttpError(Exception):
def __init__(self, status, payload):
self.status = status
self.payload = payload
Exception.__init__(self, "HTTP %s: %s" % (status, payload))
class _SocketReader:
def __init__(self, sock, initial=b""):
self.sock = sock
self.buffer = initial
def read_exact(self, length):
chunks = []
if self.buffer:
take = min(length, len(self.buffer))
chunks.append(self.buffer[:take])
self.buffer = self.buffer[take:]
length -= take
while length:
chunk = self.sock.recv(min(length, 4096))
if not chunk:
raise OSError("HTTP response ended early")
chunks.append(chunk)
length -= len(chunk)
return b"".join(chunks)
def _read_response(sock):
head = b""
while b"\r\n\r\n" not in head:
chunk = sock.recv(1024)
if not chunk:
raise OSError("HTTP response header ended early")
head += chunk
if len(head) > 16 * 1024:
raise OSError("HTTP response header is too large")
raw_head, initial = head.split(b"\r\n\r\n", 1)
lines = raw_head.decode("utf-8").split("\r\n")
parts = lines[0].split(" ", 2)
if len(parts) < 2:
raise OSError("invalid HTTP status line")
status = int(parts[1])
headers = {}
for line in lines[1:]:
if ":" not in line:
continue
name, value = line.split(":", 1)
headers[name.strip().lower()] = value.strip()
length = int(headers.get("content-length", "0"))
body = _SocketReader(sock, initial).read_exact(length) if length else b""
payload = None
if body:
content_type = headers.get("content-type", "")
if "json" in content_type:
payload = json.loads(body.decode("utf-8"))
else:
payload = body
return status, headers, payload
class LEDStatus:
"""N6 user RGB LED state helper."""
def __init__(self):
try:
from machine import LED
self.red = LED("LED_RED")
self.green = LED("LED_GREEN")
self.blue = LED("LED_BLUE")
except (ImportError, ValueError):
self.red = self.green = self.blue = None
def _set(self, red=False, green=False, blue=False):
for item, enabled in (
(self.red, red),
(self.green, green),
(self.blue, blue),
):
if item is not None:
item.on() if enabled else item.off()
def connecting(self):
self._set(blue=True)
def active(self):
self._set(green=True)
def uploading(self, alternate=False):
self._set(green=alternate, blue=not alternate)
def error(self):
self._set(red=True)
def off(self):
self._set()
def connect_wifi(ssid, password, timeout_seconds=30):
import network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
if not wlan.isconnected():
wlan.connect(ssid, password)
started = ticks_ms()
while not wlan.isconnected():
if ticks_diff(ticks_ms(), started) > timeout_seconds * 1000:
raise OSError("Wi-Fi connection timed out")
sleep_ms(250)
return wlan
class MeetingAsrClient:
def __init__(self, server_url, device_id, device_token, timeout_seconds=30):
self.server_url = server_url.rstrip("/")
self.host, self.port = _parse_http_url(self.server_url)
self.device_id = device_id
self.device_token = device_token
self.timeout_seconds = timeout_seconds
def _connect(self):
address = socket.getaddrinfo(self.host, self.port, 0, socket.SOCK_STREAM)[0][-1]
sock = socket.socket()
sock.settimeout(self.timeout_seconds)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
sock.connect(address)
return sock
def _request(self, method, path, token, body=None, headers=None):
headers = dict(headers or {})
raw_body = b""
if body is not None:
if isinstance(body, (bytes, bytearray, memoryview)):
raw_body = body
else:
raw_body = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
headers["Host"] = self.host
headers["Authorization"] = "Bearer " + token
headers["Connection"] = "close"
headers["Content-Length"] = str(len(raw_body))
request = (
"%s %s HTTP/1.1\r\n" % (method, path)
+ "".join("%s: %s\r\n" % item for item in headers.items())
+ "\r\n"
).encode("utf-8")
sock = self._connect()
try:
_safe_send(sock, request)
if raw_body:
_safe_send(sock, raw_body)
status, response_headers, payload = _read_response(sock)
finally:
sock.close()
if status < 200 or status >= 300:
raise HttpError(status, payload)
return status, response_headers, payload
def create_recorded_session(
self,
video_bytes,
audio_bytes,
max_duration_seconds,
capture,
email=None,
meeting_type=None,
):
return self._request(
"POST",
"/api/v1/device-sessions",
self.device_token,
{
"device_id": self.device_id,
"mode": "recorded",
"max_duration_seconds": max_duration_seconds,
"capture": capture,
"files": {
"video": {"size_bytes": video_bytes},
"audio": {"size_bytes": audio_bytes},
},
"email": email,
"meeting_type": meeting_type,
},
)[2]
def create_recorded_audio_session(
self,
audio_bytes,
max_duration_seconds,
capture,
email=None,
meeting_type=None,
):
return self._request(
"POST",
"/api/v1/device-sessions",
self.device_token,
{
"device_id": self.device_id,
"mode": "recorded",
"max_duration_seconds": max_duration_seconds,
"capture": capture,
"files": {
"audio": {"size_bytes": audio_bytes},
},
"email": email,
"meeting_type": meeting_type,
},
)[2]
def create_live_session(self, max_duration_seconds, capture, email=None, meeting_type=None):
return self._request(
"POST",
"/api/v1/device-sessions",
self.device_token,
{
"device_id": self.device_id,
"mode": "live",
"max_duration_seconds": max_duration_seconds,
"capture": capture,
"email": email,
"meeting_type": meeting_type,
},
)[2]
def status(self, session_id, session_token):
return self._request(
"GET",
"/api/v1/device-sessions/" + session_id,
session_token,
)[2]
def upload_offset(self, session_id, session_token, slot):
_, headers, _ = self._request(
"HEAD",
"/api/v1/device-sessions/%s/files/%s" % (session_id, slot),
session_token,
)
return int(headers["upload-offset"]), int(headers["upload-length"])
def upload_file(
self,
session_id,
session_token,
slot,
file_path,
retry_count=5,
progress=None,
):
offset, expected = self.upload_offset(session_id, session_token, slot)
local_size = __import__("os").stat(file_path)[6]
if local_size != expected:
raise OSError(
"Local recording size does not match declared upload length"
)
buffer = bytearray(UPLOAD_CHUNK_BYTES)
buffer_view = memoryview(buffer)
with open(file_path, "rb") as source:
while offset < expected:
source.seek(offset)
chunk_size = min(UPLOAD_CHUNK_BYTES, expected - offset)
actual = source.readinto(buffer_view[:chunk_size])
if not actual:
raise OSError("Local recording is shorter than declared")
body = buffer_view[:actual]
attempts = 0
while True:
try:
_, headers, _ = self._request(
"PATCH",
"/api/v1/device-sessions/%s/files/%s"
% (session_id, slot),
session_token,
body,
{
"Content-Type": "application/offset+octet-stream",
"Upload-Offset": str(offset),
},
)
offset = int(headers["upload-offset"])
break
except (OSError, HttpError):
attempts += 1
if attempts > retry_count:
raise
sleep_ms(min(8000, 500 * (2 ** (attempts - 1))))
offset, _ = self.upload_offset(
session_id, session_token, slot
)
if progress:
progress(slot, offset, expected)
return offset
def wait_for_job(
self,
session_id,
session_token,
timeout_seconds=600,
poll_seconds=2,
):
started = ticks_ms()
while ticks_diff(ticks_ms(), started) < timeout_seconds * 1000:
status = self.status(session_id, session_token)
if status.get("job_id"):
return status
if status.get("state") in ("FAILED", "INTERRUPTED", "EXPIRED"):
raise OSError("Server rejected media: %s" % status.get("error"))
time.sleep(poll_seconds)
raise OSError("Timed out waiting for server job")
def open_live_stream(self, session_id, session_token):
path = "/api/v1/device-sessions/%s/stream" % session_id
sock = self._connect()
request = (
"POST %s HTTP/1.1\r\n" % path
+ "Host: %s\r\n" % self.host
+ "Authorization: Bearer %s\r\n" % session_token
+ "Content-Type: application/x-meeting-asr-av\r\n"
+ "X-Meeting-ASR-Protocol: mav1\r\n"
+ "Transfer-Encoding: chunked\r\n"
+ "Connection: close\r\n\r\n"
).encode("utf-8")
_safe_send(sock, request)
return ChunkedAvStream(sock)
class ChunkedAvStream:
def __init__(self, sock):
self.sock = sock
self.sequence = 0
self.closed = False
def write_record(self, record_type, timestamp_ms, payload=b""):
if self.closed:
raise OSError("A/V stream is closed")
raw = av_record(record_type, self.sequence, timestamp_ms, payload)
self.sequence += 1
_safe_send(self.sock, ("%x\r\n" % len(raw)).encode("ascii"))
_safe_send(self.sock, raw)
_safe_send(self.sock, b"\r\n")
def finish(self, timestamp_ms):
if self.closed:
return None
self.write_record(AV_END, timestamp_ms)
_safe_send(self.sock, b"0\r\n\r\n")
status, headers, payload = _read_response(self.sock)
self.sock.close()
self.closed = True
if status < 200 or status >= 300:
raise HttpError(status, payload)
return payload
def abort(self):
if not self.closed:
self.sock.close()
self.closed = True
"""Record one HD AVI on OpenMV N6 and upload it on the same trusted LAN.
This is an additional workflow. It does not replace ``upload.py`` or the
existing WAV/MJPEG programs. The AVI must contain both video and 16 kHz mono
audio; the server verifies both streams before creating a job.
"""
import gc
import json
import os
import time
import csi
import network
from avi_recorder import AviWriter
from machine import LED
import device_config as config
from meeting_asr_client import MeetingAsrClient
OUTPUT_FILE = "/sdcard/hd_record.avi"
RECORD_SECS = 6
QUALITY = 70
WIDTH = 1280
HEIGHT = 720
WIFI_TIMEOUT_SECONDS = 60
SESSION_SECONDS = max(180, RECORD_SECS + 120)
# 上传时附加的两个业务参数
UPLOAD_EMAIL = "1"
UPLOAD_MEETING_TYPE = "2"
led_red = LED("LED_RED")
led_green = LED("LED_GREEN")
led_blue = LED("LED_BLUE")
