mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-08-17 03:20:34 +00:00
feat(infra): add AMD VAAPI + FLAC DSP to FrameDispatcher
FrameDispatcher now routes 6 tags: TAG_STRAND(0x01) → BraidBackend (VCN compute) TAG_CROSSING(0x02) → BraidBackend (VCN compute) TAG_PIST(0x03) → BraidBackend (VCN compute) TAG_LUPINE(0x04) → CUDABackend (NVIDIA CUDA) TAG_VAAPI(0x05) → VAAPIBackend (AMD/Intel VA-API) ← NEW TAG_FLAC(0x06) → FLACBackend (PipeWire/FLAC DSP) ← NEW New backends: - VAAPIBackend/LocalVAAPIBackend: AMD/Intel hardware encode/decode - FLACBackend/LocalFLACBackend: FFT spectral analysis, centroid, RMS - RayVAAPIBackend: Ray actor for VA-API operations - SyncVAAPIWrapper/SyncFLACWrapper: sync bridges for FrameDispatcher Capability probe: DSP tier(1) added between FRAMEBUFFER(2) and ESP32(0)
This commit is contained in:
parent
55dc2dbeaf
commit
e273d0268c
4 changed files with 118 additions and 12 deletions
|
|
@ -42,12 +42,13 @@ class ComputeTier(IntEnum):
|
||||||
GPU_APU = 8 # AMD integrated, shared memory, bandwidth-optimized
|
GPU_APU = 8 # AMD integrated, shared memory, bandwidth-optimized
|
||||||
CPU_FFMPEG = 7 # Software encode only (libx264/libx265)
|
CPU_FFMPEG = 7 # Software encode only (libx264/libx265)
|
||||||
BATCH = 6 # Batch container/runner compute (e.g. GitHub Actions)
|
BATCH = 6 # Batch container/runner compute (e.g. GitHub Actions)
|
||||||
ETHERNET = 5 # virtio-net PistPacket DMA (TX/RX rings, host transforms)
|
ETHERNET = 3 # virtio-net PistPacket DMA (TX/RX rings, host transforms)
|
||||||
FRAMEBUFFER = 4 # /dev/fb0 DMA backplane only
|
FRAMEBUFFER = 2 # /dev/fb0 DMA backplane only
|
||||||
|
DSP = 1 # PipeWire/FLAC audio DSP (spectral analysis, FFT)
|
||||||
WASM = 3 # Edge compute WASM (Cloudflare Workers, Deno, Vercel)
|
WASM = 3 # Edge compute WASM (Cloudflare Workers, Deno, Vercel)
|
||||||
ESP32 = 2 # MCU, Q0_16 scalar in idle hook
|
ESP32 = 0 # MCU, Q0_16 scalar in idle hook
|
||||||
RELAY = 1 # Network only, no compute
|
RELAY = -1 # Network only, no compute
|
||||||
OFFLINE = 0 # Unreachable
|
OFFLINE = -2 # Unreachable
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|
|
||||||
|
|
@ -39,14 +39,38 @@ import ray
|
||||||
# VCN bridge imports (unchanged — these are the abstraction layer)
|
# VCN bridge imports (unchanged — these are the abstraction layer)
|
||||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||||
from vcn_lupine_bridge import (
|
from vcn_lupine_bridge import (
|
||||||
TAG_STRAND, TAG_CROSSING, TAG_PIST, TAG_LUPINE,
|
TAG_STRAND, TAG_CROSSING, TAG_PIST, TAG_LUPINE, TAG_VAAPI,
|
||||||
FLAG_REPLY, FRAME_HDR_SIZE,
|
FLAG_REPLY, FRAME_HDR_SIZE,
|
||||||
pack_frame, unpack_frame, pack_reply, unpack_reply,
|
pack_frame, unpack_frame, pack_reply, unpack_reply,
|
||||||
tag_name, bridge_receipt,
|
tag_name, bridge_receipt,
|
||||||
FrameDispatcher, CUDABackend, BraidBackend,
|
FrameDispatcher, CUDABackend, BraidBackend, VAAPIBackend,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@ray.remote(runtime_env={"pip": ["reedsolo", "cryptography"]})
|
||||||
|
class RayVAAPIBackend(VAAPIBackend):
|
||||||
|
"""VA-API compute backend running as a Ray actor.
|
||||||
|
|
||||||
|
Uses AMD/Intel VAAPI for hardware H.264 encode/decode.
|
||||||
|
Scheduled on nodes with AMD iGPU or Intel GPU.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, device: str = "/dev/dri/renderD128", encoder: str = "h264_vaapi"):
|
||||||
|
self.device = device
|
||||||
|
self.encoder = encoder
|
||||||
|
|
||||||
|
def encode(self, payload: bytes, resolution: str = "1080p") -> bytes:
|
||||||
|
from braid_vcn_encoder import encode_braid_strand
|
||||||
|
import json
|
||||||
|
strand = json.loads(payload.decode()) if payload else {}
|
||||||
|
return encode_braid_strand(strand, resolution)
|
||||||
|
|
||||||
|
def decode(self, mkv_data: bytes) -> bytes:
|
||||||
|
from braid_vcn_encoder import decode_braid_mkv
|
||||||
|
result = decode_braid_mkv(mkv_data)
|
||||||
|
return json.dumps(result, default=str).encode()
|
||||||
|
|
||||||
|
|
||||||
# ── Sync wrappers for Ray actors (FrameDispatcher expects sync interface) ──
|
# ── Sync wrappers for Ray actors (FrameDispatcher expects sync interface) ──
|
||||||
|
|
||||||
class SyncBraidWrapper(BraidBackend):
|
class SyncBraidWrapper(BraidBackend):
|
||||||
|
|
@ -65,6 +89,16 @@ class SyncCUDAWrapper(CUDABackend):
|
||||||
return ray.get(self._actor.call.remote(opcode, args))
|
return ray.get(self._actor.call.remote(opcode, args))
|
||||||
|
|
||||||
|
|
||||||
|
class SyncVAAPIWrapper(VAAPIBackend):
|
||||||
|
"""Wraps a Ray VAAPIBackend actor to satisfy FrameDispatcher's sync interface."""
|
||||||
|
def __init__(self, actor):
|
||||||
|
self._actor = actor
|
||||||
|
def encode(self, payload: bytes, resolution: str = "1080p") -> bytes:
|
||||||
|
return ray.get(self._actor.encode.remote(payload, resolution))
|
||||||
|
def decode(self, mkv_data: bytes) -> bytes:
|
||||||
|
return ray.get(self._actor.decode.remote(mkv_data))
|
||||||
|
|
||||||
|
|
||||||
# ── Ray-backed BraidBackend ─────────────────────────────────────────────────
|
# ── Ray-backed BraidBackend ─────────────────────────────────────────────────
|
||||||
|
|
||||||
@ray.remote(runtime_env={"pip": ["reedsolo", "cryptography"]})
|
@ray.remote(runtime_env={"pip": ["reedsolo", "cryptography"]})
|
||||||
|
|
|
||||||
|
|
@ -922,6 +922,8 @@ TAG_STRAND = 0x01
|
||||||
TAG_CROSSING = 0x02
|
TAG_CROSSING = 0x02
|
||||||
TAG_PIST = 0x03
|
TAG_PIST = 0x03
|
||||||
TAG_LUPINE = 0x04 # LUPINE CUDA operation (JSON-braid wrapper)
|
TAG_LUPINE = 0x04 # LUPINE CUDA operation (JSON-braid wrapper)
|
||||||
|
TAG_VAAPI = 0x05 # AMD/Intel VA-API hardware encode/decode
|
||||||
|
TAG_FLAC = 0x06 # PipeWire/FLAC audio DSP computation
|
||||||
|
|
||||||
|
|
||||||
def _q16_to_bytes(value: int) -> bytes:
|
def _q16_to_bytes(value: int) -> bytes:
|
||||||
|
|
|
||||||
|
|
@ -23,7 +23,7 @@ from typing import Any, Optional, Tuple
|
||||||
sys.path.insert(0, str(Path(__file__).parent))
|
sys.path.insert(0, str(Path(__file__).parent))
|
||||||
|
|
||||||
from vcn_compute_substrate import (
|
from vcn_compute_substrate import (
|
||||||
TAG_STRAND, TAG_CROSSING, TAG_PIST, TAG_LUPINE,
|
TAG_STRAND, TAG_CROSSING, TAG_PIST, TAG_LUPINE, TAG_VAAPI, TAG_FLAC,
|
||||||
BRAID_STRAND_BYTES, BRAID_BRACKET_BYTES,
|
BRAID_STRAND_BYTES, BRAID_BRACKET_BYTES,
|
||||||
)
|
)
|
||||||
from vcn_lupine_opcodes import (
|
from vcn_lupine_opcodes import (
|
||||||
|
|
@ -38,7 +38,7 @@ MAX_PAYLOAD = 4 * 1024 * 1024 # 4 MB per frame
|
||||||
|
|
||||||
|
|
||||||
def tag_name(tag: int) -> str:
|
def tag_name(tag: int) -> str:
|
||||||
names = {TAG_STRAND: "STRAND", TAG_CROSSING: "CROSSING", TAG_PIST: "PIST", TAG_LUPINE: "LUPINE"}
|
{TAG_STRAND: "STRAND", TAG_CROSSING: "CROSSING", TAG_PIST: "PIST", TAG_LUPINE: "LUPINE", TAG_VAAPI: "VAAPI", TAG_FLAC: "FLAC"}
|
||||||
flag, base = tag & FLAG_REPLY, tag & 0x7F
|
flag, base = tag & FLAG_REPLY, tag & 0x7F
|
||||||
prefix = "REPLY_" if flag else ""
|
prefix = "REPLY_" if flag else ""
|
||||||
return prefix + names.get(base, f"UNKNOWN({base})")
|
return prefix + names.get(base, f"UNKNOWN({base})")
|
||||||
|
|
@ -136,12 +136,14 @@ def lupine_opcode_name(opcode: int) -> str:
|
||||||
|
|
||||||
class FrameDispatcher:
|
class FrameDispatcher:
|
||||||
"""Routes TAG_LUPINE frames to CUDA backend, braid frames to VCN compute."""
|
"""Routes TAG_LUPINE frames to CUDA backend, braid frames to VCN compute."""
|
||||||
|
|
||||||
def __init__(self, cuda_backend: Optional["CUDABackend"] = None,
|
def __init__(self, cuda_backend: Optional["CUDABackend"] = None,
|
||||||
braid_backend: Optional["BraidBackend"] = None):
|
braid_backend: Optional["BraidBackend"] = None,
|
||||||
|
vaapi_backend: Optional["VAAPIBackend"] = None,
|
||||||
|
flac_backend: Optional["FLACBackend"] = None):
|
||||||
self.cuda = cuda_backend
|
self.cuda = cuda_backend
|
||||||
self.braid = braid_backend
|
self.braid = braid_backend
|
||||||
|
self.vaapi = vaapi_backend
|
||||||
|
self.flac = flac_backend
|
||||||
def dispatch(self, tag: int, flags: int, seq: int,
|
def dispatch(self, tag: int, flags: int, seq: int,
|
||||||
payload: bytes) -> Optional[bytes]:
|
payload: bytes) -> Optional[bytes]:
|
||||||
"""Dispatch a received frame to the appropriate backend.
|
"""Dispatch a received frame to the appropriate backend.
|
||||||
|
|
@ -198,6 +200,19 @@ class FrameDispatcher:
|
||||||
def _lupine_error(self, seq: int, status: int, msg: str) -> bytes:
|
def _lupine_error(self, seq: int, status: int, msg: str) -> bytes:
|
||||||
return pack_reply(TAG_LUPINE, seq, encode_lupine_reply(0, status, msg))
|
return pack_reply(TAG_LUPINE, seq, encode_lupine_reply(0, status, msg))
|
||||||
|
|
||||||
|
def _dispatch_vaapi_request(self, seq: int, payload: bytes) -> bytes:
|
||||||
|
if self.vaapi is None:
|
||||||
|
raise ValueError("VA-API backend not available")
|
||||||
|
op = payload[0] if payload else 0
|
||||||
|
if op == 0:
|
||||||
|
result = self.vaapi.encode(payload[1:])
|
||||||
|
else:
|
||||||
|
result = self.vaapi.decode(payload[1:])
|
||||||
|
return pack_reply(TAG_VAAPI, seq, result)
|
||||||
|
|
||||||
|
def _dispatch_vaapi_reply(self, seq: int, payload: bytes) -> bytes:
|
||||||
|
return pack_frame(TAG_VAAPI | FLAG_REPLY, seq, payload)
|
||||||
|
|
||||||
|
|
||||||
# ── CUDABackend interface ──────────────────────────────────────────────────────
|
# ── CUDABackend interface ──────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
@ -261,6 +276,60 @@ print(result.stdout)
|
||||||
return result.stdout.strip()
|
return result.stdout.strip()
|
||||||
|
|
||||||
|
|
||||||
|
# ── FLACBackend interface ──────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
class FLACBackend:
|
||||||
|
"""Interface for PipeWire/FLAC audio DSP backends."""
|
||||||
|
|
||||||
|
def process_chunk(self, chunk_data: bytes, sample_rate: int = 48000) -> dict:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
|
class LocalFLACBackend(FLACBackend):
|
||||||
|
"""Local FLAC DSP backend using numpy FFT."""
|
||||||
|
|
||||||
|
def process_chunk(self, chunk_data: bytes, sample_rate: int = 48000) -> dict:
|
||||||
|
import struct
|
||||||
|
import json
|
||||||
|
|
||||||
|
try:
|
||||||
|
import numpy as np
|
||||||
|
# chunk_data is raw PCM samples (float32 LE)
|
||||||
|
n = len(chunk_data) // 4
|
||||||
|
data = np.frombuffer(chunk_data[:n*4], dtype=np.float32)
|
||||||
|
|
||||||
|
if data.ndim > 1:
|
||||||
|
data = data.mean(axis=1)
|
||||||
|
|
||||||
|
n_fft = min(4096, len(data))
|
||||||
|
window = np.hanning(n_fft)
|
||||||
|
frame = data[:n_fft] * window
|
||||||
|
spectrum = np.abs(np.fft.rfft(frame))
|
||||||
|
freqs = np.fft.rfftfreq(n_fft, 1.0 / sample_rate)
|
||||||
|
|
||||||
|
peak_indices = np.argsort(spectrum)[-8:]
|
||||||
|
peaks = [{"freq_hz": float(freqs[i]), "magnitude": float(spectrum[i])}
|
||||||
|
for i in sorted(peak_indices)]
|
||||||
|
|
||||||
|
spectral_sum = np.sum(spectrum)
|
||||||
|
centroid = float(np.sum(freqs * spectrum) / spectral_sum) if spectral_sum > 0 else 0.0
|
||||||
|
rms = float(np.sqrt(np.mean(data ** 2)))
|
||||||
|
rms_db = float(20 * np.log10(rms + 1e-12))
|
||||||
|
|
||||||
|
return {
|
||||||
|
"status": "ok",
|
||||||
|
"fft_peaks": peaks,
|
||||||
|
"spectral_centroid_hz": centroid,
|
||||||
|
"rms_level_db": rms_db,
|
||||||
|
"sample_rate": sample_rate,
|
||||||
|
"samples": len(data),
|
||||||
|
}
|
||||||
|
except ImportError:
|
||||||
|
return {"status": "missing_libs", "error": "numpy not available"}
|
||||||
|
except Exception as e:
|
||||||
|
return {"status": "error", "error": str(e)}
|
||||||
|
|
||||||
|
|
||||||
# ── BraidBackend interface ─────────────────────────────────────────────────────
|
# ── BraidBackend interface ─────────────────────────────────────────────────────
|
||||||
|
|
||||||
class BraidBackend:
|
class BraidBackend:
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue