#!/usr/bin/env python3 """Wyoming lazy proxy for faster-whisper. Home Assistant talks to this proxy instead of faster-whisper directly. - `describe` events are answered from a cached `info` response, so HA's availability checks never touch the backend. - The first *real* event of a connection (e.g. `transcribe`) starts the faster-whisper container via the docker socket, replays the event, and relays bytes in both directions from then on. - After IDLE_SECONDS without an active session the container is stopped again, freeing its VRAM. stdlib only; runs on any python:3.12 image. """ import asyncio import json import os import socket import time BACKEND_HOST = os.environ.get('BACKEND_HOST', '10.0.0.4') BACKEND_PORT = int(os.environ.get('BACKEND_PORT', '10302')) PROXY_PORT = int(os.environ.get('PROXY_PORT', '10300')) IDLE_SECONDS = int(os.environ.get('IDLE_SECONDS', '900')) CONTAINER = os.environ.get('CONTAINER', 'faster-whisper') DOCKER_SOCK = os.environ.get('DOCKER_SOCK', '/var/run/docker.sock') CACHE_FILE = os.environ.get('CACHE_FILE', '/app/info.cache') START_TIMEOUT = int(os.environ.get('START_TIMEOUT', '180')) state = {'active': 0, 'last': time.monotonic(), 'lock': None} def log(*args): print(time.strftime('%H:%M:%S'), *args, flush=True) def docker_api(method, path): """Minimal docker engine API call over the unix socket. Returns (status, body).""" sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(60) sock.connect(DOCKER_SOCK) sock.sendall(f'{method} {path} HTTP/1.1\r\nHost: docker\r\nConnection: close\r\n\r\n'.encode()) buf = b'' while True: chunk = sock.recv(65536) if not chunk: break buf += chunk sock.close() head, _, body = buf.partition(b'\r\n\r\n') return int(head.split(b' ', 2)[1]), body def backend_running(): try: status, body = docker_api('GET', f'/containers/{CONTAINER}/json') if status != 200: return False text = body.decode('utf-8', 'replace') text = text[text.find('{'):text.rfind('}') + 1] return json.loads(text).get('State', {}).get('Running', False) except Exception as err: log('backend_running check failed:', err) return False def start_container(): status, _ = docker_api('POST', f'/containers/{CONTAINER}/start') return status in (204, 304) def stop_container(): status, _ = docker_api('POST', f'/containers/{CONTAINER}/stop?t=30') return status in (204, 304) async def read_event(reader): """Reads one wyoming event; returns {'raw': bytes, 'type': str} or None on EOF.""" line = await reader.readline() if not line: return None header = json.loads(line) extra = b'' for key in ('data_length', 'payload_length'): length = int(header.get(key) or 0) if length: extra += await reader.readexactly(length) return {'raw': line + extra, 'type': header.get('type')} async def ensure_backend(): """Starts the backend container if needed and returns an open (reader, writer). A bare TCP connect is not proof of readiness: docker-proxy accepts on the published port as soon as the container starts, long before the Wyoming server listens. So readiness = a real describe -> info handshake. """ async with state['lock']: deadline = time.monotonic() + START_TIMEOUT if not await asyncio.to_thread(backend_running): log('starting backend container') await asyncio.to_thread(start_container) while time.monotonic() < deadline: try: reader, writer = await asyncio.open_connection(BACKEND_HOST, BACKEND_PORT) writer.write(json.dumps({'type': 'describe', 'data': {}}).encode() + b'\n') await writer.drain() event = await asyncio.wait_for(read_event(reader), 5) if event and event['type'] == 'info': # Connection is clean again after consuming the info event. return reader, writer writer.close() except (OSError, asyncio.TimeoutError, asyncio.IncompleteReadError, json.JSONDecodeError): pass await asyncio.sleep(1) raise RuntimeError('backend did not come up within START_TIMEOUT') async def refresh_cache(): try: reader, writer = await asyncio.open_connection(BACKEND_HOST, BACKEND_PORT) writer.write(json.dumps({'type': 'describe', 'data': {}}).encode() + b'\n') await writer.drain() event = await asyncio.wait_for(read_event(reader), 20) writer.close() if event and event['type'] == 'info': with open(CACHE_FILE, 'wb') as f: f.write(event['raw']) log('info cache refreshed:', len(event['raw']), 'bytes') except Exception as err: log('cache refresh failed:', err) async def pipe(reader, writer): try: while True: chunk = await reader.read(65536) if not chunk: break writer.write(chunk) await writer.drain() except (ConnectionResetError, BrokenPipeError): pass finally: try: writer.close() except Exception: pass async def handle_client(client_reader, client_writer): peer = client_writer.get_extra_info('peername') backend = None try: while backend is None: event = await read_event(client_reader) if event is None: return if event['type'] == 'ping': client_writer.write(json.dumps({'type': 'pong', 'data': {}}).encode() + b'\n') await client_writer.drain() continue if event['type'] == 'describe' and os.path.exists(CACHE_FILE): with open(CACHE_FILE, 'rb') as f: client_writer.write(f.read()) await client_writer.drain() continue # Real work (or describe without a cache yet): wake the backend. state['active'] += 1 state['last'] = time.monotonic() try: backend = await ensure_backend() except Exception as err: state['active'] -= 1 state['last'] = time.monotonic() log('backend wake failed:', err) return log(f'session start from {peer} (trigger: {event["type"]})') backend[1].write(event['raw']) await backend[1].drain() asyncio.ensure_future(refresh_cache()) await asyncio.gather(pipe(client_reader, backend[1]), pipe(backend[0], client_writer)) except (ConnectionResetError, BrokenPipeError, asyncio.IncompleteReadError, json.JSONDecodeError) as err: log('client error:', err) finally: if backend is not None: state['active'] -= 1 state['last'] = time.monotonic() log('session end, active:', state['active']) try: client_writer.close() except Exception: pass async def idle_reaper(): while True: await asyncio.sleep(30) idle = time.monotonic() - state['last'] if state['active'] == 0 and idle > IDLE_SECONDS and await asyncio.to_thread(backend_running): log(f'idle for {int(idle)}s - stopping backend container') await asyncio.to_thread(stop_container) async def main(): state['lock'] = asyncio.Lock() if not os.path.exists(CACHE_FILE): log('no info cache - bootstrapping backend once to capture it') try: reader, writer = await ensure_backend() writer.close() await refresh_cache() except Exception as err: log('bootstrap failed (will retry on first request):', err) server = await asyncio.start_server(handle_client, '0.0.0.0', PROXY_PORT) log(f'wyoming lazy proxy listening on :{PROXY_PORT} -> {BACKEND_HOST}:{BACKEND_PORT} ' f'(container {CONTAINER}, idle {IDLE_SECONDS}s)') async with server: await asyncio.gather(server.serve_forever(), idle_reaper()) if __name__ == '__main__': asyncio.run(main())