"""Binary serialization of Draw2D batches for WebSocket streaming.
Serializes the output of ``Draw2D._get_batches()`` into compact binary frames
for transmission to a browser-side WebGPU 2D renderer.
Wire format per frame::
Header: frame_id(u32) + batch_count(u32) + atlas_version(u32) [12 bytes]
Per batch:
clip_rect(4×f32) [16 bytes]
sub-batches (variable), terminated by 0xFF sentinel byte
Per sub-batch:
draw_type(u8) + vertex_count(u32) + index_count(u32) [9 bytes]
raw vertex bytes (vertex_count × 32)
raw index bytes (index_count × 4)
Draw types: 0=fill, 1=line, 2=text, 3=textured (textured adds texture_id u32 before vertices).
"""
from __future__ import annotations
import struct
from typing import Any
import numpy as np
__all__ = ["DrawSerializer"]
# Draw type constants
DRAW_FILL = 0
DRAW_LINE = 1
DRAW_TEXT = 2
DRAW_TEXTURED = 3
BATCH_END = 0xFF
# Header: frame_id(u32) + batch_count(u32) + atlas_version(u32)
_HEADER = struct.Struct("<III")
# Clip rect: x, y, w, h as f32
_CLIP = struct.Struct("<ffff")
# Sub-batch header: draw_type(u8) + vertex_count(u32) + index_count(u32)
_SUB_HEADER = struct.Struct("<BII")
# Texture ID prefix for textured sub-batches
_TEX_ID = struct.Struct("<I")
[docs]
class DrawSerializer:
"""Serializes Draw2D batch output into binary frames for WebSocket streaming."""
[docs]
@staticmethod
def serialize_frame(frame_id: int, batches: list, atlas_version: int = 0) -> bytes:
"""Serialize a list of Draw2D batches into a binary frame.
Args:
frame_id: Monotonic frame counter.
batches: Output of ``Draw2D._get_batches()`` — list of
(clip, fill_data, line_data, text_data, tex_draws).
atlas_version: Current MSDF atlas version for cache invalidation.
Returns:
Compact binary frame ready for WebSocket transmission.
"""
parts: list[bytes] = []
parts.append(_HEADER.pack(frame_id, len(batches), atlas_version))
for batch in batches:
clip, fill_data, line_data, text_data, tex_draws = batch
# Clip rect (0,0,0,0 means no clip / full viewport)
if clip is not None:
parts.append(_CLIP.pack(float(clip[0]), float(clip[1]), float(clip[2]), float(clip[3])))
else:
parts.append(_CLIP.pack(0.0, 0.0, 0.0, 0.0))
# Fill triangles
if fill_data is not None:
verts, indices = fill_data
_append_sub_batch(parts, DRAW_FILL, verts, indices)
# Lines
if line_data is not None:
# line_data is just a vertex array (no indices — line-list)
verts = line_data
_append_sub_batch(parts, DRAW_LINE, verts, None)
# Text (MSDF)
if text_data is not None:
verts, indices = text_data
_append_sub_batch(parts, DRAW_TEXT, verts, indices)
# Textured quads (per-texture draws)
for tex_draw in tex_draws:
tex_id, verts, indices = tex_draw
parts.append(_SUB_HEADER.pack(DRAW_TEXTURED, len(verts), len(indices)))
parts.append(_TEX_ID.pack(tex_id))
parts.append(verts.tobytes())
parts.append(indices.tobytes())
# Batch terminator
parts.append(bytes([BATCH_END]))
return b"".join(parts)
[docs]
@staticmethod
def deserialize_frame(data: bytes) -> dict[str, Any]:
"""Deserialize a binary frame back into structured data (for testing).
Returns:
Dict with keys: frame_id, atlas_version, batches.
Each batch: dict with clip, sub_batches list.
Each sub_batch: dict with draw_type, vertices, indices, texture_id (optional).
"""
from ..draw2d import UI_VERTEX_DTYPE
offset = 0
frame_id, batch_count, atlas_version = _HEADER.unpack_from(data, offset)
offset += _HEADER.size
batches = []
for _ in range(batch_count):
clip = _CLIP.unpack_from(data, offset)
offset += _CLIP.size
clip = None if clip == (0.0, 0.0, 0.0, 0.0) else clip
sub_batches = []
while offset < len(data):
tag = data[offset]
if tag == BATCH_END:
offset += 1
break
draw_type, vert_count, idx_count = _SUB_HEADER.unpack_from(data, offset)
offset += _SUB_HEADER.size
texture_id = None
if draw_type == DRAW_TEXTURED:
texture_id = _TEX_ID.unpack_from(data, offset)[0]
offset += _TEX_ID.size
vert_bytes = vert_count * UI_VERTEX_DTYPE.itemsize
verts = np.frombuffer(data[offset : offset + vert_bytes], dtype=UI_VERTEX_DTYPE).copy()
offset += vert_bytes
indices = None
if idx_count > 0:
idx_bytes = idx_count * 4
indices = np.frombuffer(data[offset : offset + idx_bytes], dtype=np.uint32).copy()
offset += idx_bytes
sb: dict[str, Any] = {"draw_type": draw_type, "vertices": verts, "indices": indices}
if texture_id is not None:
sb["texture_id"] = texture_id
sub_batches.append(sb)
batches.append({"clip": clip, "sub_batches": sub_batches})
return {"frame_id": frame_id, "atlas_version": atlas_version, "batches": batches}
def _append_sub_batch(parts: list[bytes], draw_type: int, verts: np.ndarray, indices: np.ndarray | None) -> None:
"""Append a sub-batch (header + raw vertex/index bytes) to the parts list."""
idx_count = len(indices) if indices is not None else 0
parts.append(_SUB_HEADER.pack(draw_type, len(verts), idx_count))
parts.append(verts.tobytes())
if indices is not None:
parts.append(indices.tobytes())