"""Frame types and pixel-format helpers.
:class:`Frame` (a decoded or fd-backed video frame), :class:`FrameHandle`
(retained dma-buf planes) and the pixel-format tables shared by the media
clients and the DSP path. Split out of media.py; media.py re-exports the
public surface for backwards compatibility.
"""
from __future__ import annotations
import fcntl
import logging
import mmap
import os
import struct
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Any, Callable
import numpy as np
from ._fallbacks import warn_numpy_fallback
logger = logging.getLogger("neoruntime_ipc_sdk.frame")
__all__ = [
"Frame",
"FrameHandle",
"PixelFormat",
"PIXEL_FORMAT_NAMES",
"StreamInfo",
]
PIXEL_FORMAT_NAMES = {
PixelFormat.NV12: "NV12",
PixelFormat.NV21: "NV21",
PixelFormat.RGB: "RGB",
PixelFormat.BGR: "BGR",
PixelFormat.RGBA: "RGBA",
PixelFormat.BGRA: "BGRA",
PixelFormat.GRAY8: "GRAY8",
PixelFormat.YUYV: "YUYV",
}
# Formats stored as plain pixel arrays (crop/resize via numpy slicing)
_PACKED_FORMATS = ("RGB", "BGR", "RGBA", "BGRA", "GRAY8")
# Planar/semi-planar YUV formats (crop/resize with per-plane handling)
_YUV_FORMATS = ("NV12", "NV21")
# Formats Frame.resize can hand to the DSP as a zero-copy source
# (must stay in sync with dsp._FRAME_FMT_TO_DSP)
_DSP_RESIZE_FORMATS = ("NV12", "RGB", "BGR", "GRAY8")
def _resize_array(img: np.ndarray, width: int, height: int) -> np.ndarray:
"""Resize a uint8 2D/3D array to (height, width).
Uses cv2 when available (INTER_AREA down / INTER_LINEAR up); falls back
to pure-numpy nearest-neighbour indexing so the SDK works without cv2.
"""
if img.shape[0] == height and img.shape[1] == width:
return img
try:
import cv2
interp = (
cv2.INTER_AREA if (height < img.shape[0] and width < img.shape[1]) else cv2.INTER_LINEAR
)
return cv2.resize(img, (width, height), interpolation=interp)
except ImportError:
warn_numpy_fallback("resize")
rows = np.arange(height) * img.shape[0] // height
cols = np.arange(width) * img.shape[1] // width
if img.ndim == 2:
return img[rows[:, None], cols[None, :]]
return img[rows[:, None], cols[None, :], :]
def _even(value: int) -> int:
"""Round down to the nearest even number, minimum 2 (YUV plane safety)."""
return max(2, value - (value % 2))
def _resize_geometry(
sw: int, sh: int, dw: int, dh: int, mode: str, yuv: bool
) -> tuple[int, int, int, int]:
"""Content box (rw, rh) and placement (ox, oy) for the resize modes.
Single source of geometry truth for the DSP fast path and both CPU
paths — the legs must agree on content placement or the same frame
would place pixels differently depending on which leg served it.
stretch: content fills dst exactly. letterbox: content is scaled to
(rw, rh) and placed at (ox, oy) inside the dst canvas. crop: content
is scaled to cover at (rw, rh) and a dst-sized window is read out of
it at (ox, oy).
"""
if mode == "stretch":
return dw, dh, 0, 0
if mode == "letterbox":
scale = min(dw / sw, dh / sh)
rw = max(1, int(round(sw * scale)))
rh = max(1, int(round(sh * scale)))
if yuv:
rw, rh = _even(rw), _even(rh)
ox, oy = (dw - rw) // 2, (dh - rh) // 2
if yuv:
ox, oy = ox & ~1, oy & ~1 # chroma-aligned pad offsets
return rw, rh, ox, oy
# crop: scale to cover, read a dst-sized window out of the middle
scale = max(dw / sw, dh / sh)
rw = max(dw, int(round(sw * scale)))
rh = max(dh, int(round(sh * scale)))
if yuv:
rw, rh = _even(rw), _even(rh)
ox, oy = (rw - dw) // 2, (rh - dh) // 2
return rw, rh, ox, oy
def _resize_transform(
src_size: tuple[int, int],
dst_size: tuple[int, int],
mode: str,
rw: int,
rh: int,
ox: int,
oy: int,
) -> dict[str, Any]:
"""Affine mapping metadata for :meth:`Frame.resize`.
Semantics (per axis): ``dst = src * scale + origin``, so the inverse —
what postprocessing needs to map model-input coordinates back to the
source frame — is ``src = (dst - origin) / scale``. For letterbox the
origin is the pad offset (content placed at (ox, oy) in the canvas);
for crop it is negative (the canvas is a window into the content).
"""
sw, sh = src_size
dw, dh = dst_size
if mode == "stretch":
return {
"op": "resize",
"src_size": (sw, sh),
"dst_size": (dw, dh),
"mode": mode,
"scale": (dw / sw, dh / sh),
"origin": (0, 0),
}
if mode == "letterbox":
return {
"op": "resize",
"src_size": (sw, sh),
"dst_size": (dw, dh),
"mode": mode,
"scale": (rw / sw, rh / sh),
"origin": (ox, oy),
}
# crop: dst pixel (x, y) shows scaled content at (x + ox, y + oy)
return {
"op": "resize",
"src_size": (sw, sh),
"dst_size": (dw, dh),
"mode": mode,
"scale": (rw / sw, rh / sh),
"origin": (-ox, -oy),
}
def _compose_transform(prev: dict[str, Any], cur: dict[str, Any]) -> dict[str, Any]:
"""Compose chained transforms: prev maps src→parent, cur maps parent→dst."""
composed = dict(cur)
composed["src_size"] = prev["src_size"]
composed["scale"] = (
prev["scale"][0] * cur["scale"][0],
prev["scale"][1] * cur["scale"][1],
)
composed["origin"] = (
prev["origin"][0] * cur["scale"][0] + cur["origin"][0],
prev["origin"][1] * cur["scale"][1] + cur["origin"][1],
)
return composed
def _frame_transform_metadata(parent_metadata: dict[str, Any], transform: dict[str, Any]) -> dict[str, Any]:
"""Copy ``parent_metadata`` and set/chain the affine ``transform`` entry."""
metadata = dict(parent_metadata)
prev = metadata.get("transform")
if isinstance(prev, dict) and "scale" in prev and "origin" in prev:
transform = _compose_transform(prev, transform)
metadata["transform"] = transform
return metadata
def _encode_jpeg(rgb: np.ndarray, quality: int = 85) -> bytes:
"""Encode an RGB uint8 array as JPEG bytes. cv2 first, Pillow fallback."""
try:
import cv2
bgr = np.ascontiguousarray(rgb[:, :, ::-1])
ok, buf = cv2.imencode(".jpg", bgr, [int(cv2.IMWRITE_JPEG_QUALITY), int(quality)])
if not ok:
raise OSError("cv2.imencode failed to encode JPEG")
return buf.tobytes()
except ImportError:
warn_numpy_fallback("JPEG encode")
import io
from PIL import Image
out = io.BytesIO()
Image.fromarray(rgb, mode="RGB").save(out, format="JPEG", quality=int(quality))
return out.getvalue()
[docs]
class FrameHandle:
"""Retained dma-buf backing store for one received frame (SDK-1).
In keep-fd mode the per-plane dma-buf fds the daemon passed with the
FRAME message are kept open here and the RELEASE message — the
daemon's buffer-recycling ticket — is deferred until :meth:`close`.
CPU access to the pixels must go through :meth:`Frame.to_array`,
which applies the required DMA_BUF_IOCTL_SYNC read fences. The fds
can also be handed to a :class:`~neoruntime_ipc_sdk.dsp.DspClient`
job as-is for a zero-copy hardware path (``resize_hw(frame, ...)``);
the geometry carried here is what the daemon needs to import them.
"""
[docs]
def __init__(
self,
fds: list[int],
strides,
plane_sizes,
frame_id: int,
on_release: Callable[[FrameHandle], None] | None = None,
width: int = 0,
height: int = 0,
format: str = "",
flags: int = 0,
):
self.fds = list(fds)
self.strides = tuple(strides)
self.plane_sizes = tuple(plane_sizes)
self.frame_id = frame_id
self.width = width
self.height = height
self.format = format
self.flags = flags
self._on_release = on_release
self._closed = False
@property
def closed(self) -> bool:
"""True once :meth:`close` ran (fds gone, frame released)."""
return self._closed
@property
def fd(self) -> int:
"""Convenience accessor for the first plane's fd."""
return self.fds[0]
[docs]
def close(self) -> None:
"""Close the fds and send the deferred RELEASE (idempotent)."""
if self._closed:
return
self._closed = True
for fd in self.fds:
try:
os.close(fd)
except OSError:
pass
if self._on_release is not None:
try:
self._on_release(self)
except Exception:
logger.exception("FrameHandle: release callback failed")
def __del__(self):
try:
self.close()
except Exception:
pass
def __repr__(self) -> str:
return (
f"FrameHandle(frame_id={self.frame_id}, "
f"{self.width}x{self.height} {self.format}, "
f"fds={len(self.fds)}, released={self._closed})"
)
# Frame metadata flags — mirror of FD_PUB_FRAME_FLAG_* in the daemon's
# fd_protocol.h. Coarse "bake active" truth per stream at dispatch time:
# a set bit means the corresponding pass ran on the dispatching path, so
# the pixels may be modified. Clean-capture apps should subscribe a
# stream outside the bake set (ai_overlay.stream_map) rather than rely
# on the bit being clear while the pass is enabled.
FRAME_FLAG_OVERLAY_BAKED = 0x1
FRAME_FLAG_DPM_BAKED = 0x2
[docs]
@dataclass
class Frame:
sequence: int
timestamp_ns: int
width: int
height: int
format: str
image: np.ndarray | None
metadata: dict[str, Any] = field(default_factory=dict)
handle: FrameHandle | None = None
flags: int = 0
@property
def baked_overlay(self) -> bool:
"""True when the AI-overlay bake pass is active on this stream."""
return bool(self.flags & FRAME_FLAG_OVERLAY_BAKED)
@property
def baked_dpm(self) -> bool:
"""True when the DPM (privacy mask) render pass is active."""
return bool(self.flags & FRAME_FLAG_DPM_BAKED)
@property
def data(self) -> np.ndarray | None:
"""Alias for image, returns raw frame data as flat numpy array."""
if self.image is None:
if self.handle is None:
return None
return self.to_array().flatten()
return self.image.flatten()
[docs]
def to_array(self) -> np.ndarray:
"""Return the raw pixel buffer, materializing a retained fd.
For keep-fd frames this maps the dma-buf planes once (with the
DMA_BUF_IOCTL_SYNC read fences) and caches the copy in
``image``; later calls return the cache without re-mapping.
"""
if self.image is None:
if self.handle is None:
raise ValueError("Frame has no image data and no retained fd")
self.image = _decode_raw(
_materialize_handle(self.handle), self.width, self.height, self.format
)
return self.image
[docs]
def release(self) -> None:
"""Release a retained fd frame back to the daemon (idempotent).
No-op for frames that were copied on receive. Also invoked by
``with frame:`` on exit — the context manager makes the keep-fd
ownership window explicit:
>>> with media.get_frame("sub", keep_fd=True) as frame:
... result = pipeline.run(frame)
>>> # handle returned to the daemon here, even on exceptions
"""
if self.handle is not None:
self.handle.close()
def __enter__(self) -> Frame:
return self
def __exit__(self, *exc) -> None:
self.release()
[docs]
def to_rgb(self) -> np.ndarray:
arr = self.to_array()
if self.format == "RGB":
return arr
elif self.format == "BGR":
return arr[:, :, ::-1]
elif self.format == "NV12":
return self._nv12_to_rgb()
elif self.format == "GRAY8":
return np.stack([arr] * 3, axis=-1)
else:
raise ValueError(f"Unsupported format: {self.format}")
def _nv12_to_rgb(self) -> np.ndarray:
try:
import cv2
return cv2.cvtColor(self.image, cv2.COLOR_YUV2RGB_NV12)
except ImportError:
warn_numpy_fallback("NV12->RGB conversion")
return self._nv12_to_rgb_pure()
def _nv12_to_rgb_pure(self) -> np.ndarray:
h, w = self.height, self.width
y = self.image[:h, :].astype(np.float32)
uv = self.image[h:, :].reshape(h // 2, w // 2, 2)
u, v = uv[:, :, 0], uv[:, :, 1]
u = u.repeat(2, axis=0).repeat(2, axis=1)
v = v.repeat(2, axis=0).repeat(2, axis=1)
y = y - 16
u = u - 128
v = v - 128
r = np.clip(1.164 * y + 1.596 * v, 0, 255).astype(np.uint8)
g = np.clip(1.164 * y - 0.813 * v - 0.391 * u, 0, 255).astype(np.uint8)
b = np.clip(1.164 * y + 2.018 * u, 0, 255).astype(np.uint8)
return np.stack([r, g, b], axis=-1)
[docs]
def crop(self, x: int, y: int, width: int, height: int) -> Frame:
"""Return a new Frame cropped to the given pixel rectangle.
NV12/NV21 require even x, y, width, height (chroma subsampling).
The original Frame is left untouched. The crop geometry is
recorded in the new frame's ``metadata['transform']`` (affine,
scale 1:1, origin (-x, -y)).
"""
if width <= 0 or height <= 0:
raise ValueError("crop width/height must be positive")
if x < 0 or y < 0 or x + width > self.width or y + height > self.height:
raise ValueError(
f"crop ({x},{y},{width}x{height}) out of bounds for "
f"{self.width}x{self.height} frame"
)
fmt = self.format
arr = self.to_array()
if fmt in _PACKED_FORMATS:
sub = np.ascontiguousarray(arr[y : y + height, x : x + width])
elif fmt in _YUV_FORMATS:
if x % 2 or y % 2 or width % 2 or height % 2:
raise ValueError(f"{fmt} crop requires even x, y, width, height")
y_plane = arr[: self.height]
uv_plane = arr[self.height :]
new_y = y_plane[y : y + height, x : x + width]
new_uv = uv_plane[y // 2 : (y + height) // 2, (x // 2) * 2 : (x // 2 + width // 2) * 2]
sub = np.ascontiguousarray(np.vstack([new_y, new_uv]))
else:
raise ValueError(f"crop not supported for format: {fmt}")
transform = {
"op": "crop",
"src_size": (self.width, self.height),
"dst_size": (width, height),
"mode": "crop",
"scale": (1.0, 1.0),
"origin": (-x, -y),
}
return Frame(
sequence=self.sequence,
timestamp_ns=self.timestamp_ns,
width=width,
height=height,
format=fmt,
image=sub,
metadata=_frame_transform_metadata(self.metadata, transform),
flags=self.flags,
)
[docs]
def resize(
self, width: int, height: int, mode: str = "letterbox", pad_value: int = 114
) -> Frame:
"""Return a new Frame resized to width x height.
Modes:
- "letterbox": fit inside, preserve aspect ratio, pad with
pad_value (NV12 pads luma with pad_value and chroma with
neutral 128). Default.
- "stretch": fill exactly, aspect ratio not preserved.
- "crop": scale to cover, center-crop the overflow.
NV12/NV21 require even target dimensions. Frames received with
keep_fd=True are scaled on the DSP without materializing their
dma-bufs first (falling back to the CPU path when the DSP
service is unavailable). The fast path respects the accel
router's policy: ``SOFTWARE_ONLY`` skips the DSP attempt,
``HARDWARE_ONLY`` raises on a failed attempt instead of
silently degrading, and the default ``PREFER_HARDWARE`` records
the fallback in the router's health counters. cv2 accelerates
the CPU path when available; a pure-numpy nearest-neighbour
path is the fallback.
The geometry is recorded in the new frame's
``metadata['transform']`` as a per-axis affine mapping
``dst = src * scale + origin`` (composed across chained
crop/resize calls). Postprocessing maps model-input coordinates
back to the source frame with ``src = (dst - origin) / scale`` —
no hand-derived letterbox maths in the app.
"""
if width <= 0 or height <= 0:
raise ValueError("resize width/height must be positive")
if mode not in ("letterbox", "stretch", "crop"):
raise ValueError(f"unsupported resize mode: {mode}")
fmt = self.format
if fmt in _YUV_FORMATS and (width % 2 or height % 2):
raise ValueError(f"{fmt} resize requires even width/height")
yuv = fmt in _YUV_FORMATS
rw, rh, ox, oy = _resize_geometry(self.width, self.height, width, height, mode, yuv)
image = self._hw_resize(width, height, mode, pad_value)
if image is None:
self.to_array() # materialize a retained fd before slicing planes
if yuv:
image = self._resize_yuv(width, height, mode, pad_value)
elif fmt in _PACKED_FORMATS:
image = self._resize_packed(width, height, mode, pad_value)
else:
raise ValueError(f"resize not supported for format: {fmt}")
transform = _resize_transform(
(self.width, self.height), (width, height), mode, rw, rh, ox, oy
)
return Frame(
sequence=self.sequence,
timestamp_ns=self.timestamp_ns,
width=width,
height=height,
format=fmt,
image=image,
metadata=_frame_transform_metadata(self.metadata, transform),
flags=self.flags,
)
def _hw_resize(self, dw: int, dh: int, mode: str, pad: int) -> np.ndarray | None:
"""Zero-copy DSP scale for keep-fd frames; None means "use CPU".
Returns the resized pixels when the frame carries a dma-buf
handle the daemon can import, or None for every other case (no
handle, closed handle, format the DSP cannot take, service
down) so :meth:`resize` can fall back to its CPU path.
Every mode scales on the DSP into the exact box the CPU path
computes, then pads/crops on the CPU, so both paths agree on
content placement:
* "stretch" — one DSP stretch to the target, used as-is.
* "letterbox" — DSP stretch to the fitted box; CPU pads with
``pad``. (The daemon's own letterbox pads Y=U=V=0 — green in
YUV, not a neutral pad — so it is never used here.)
* "crop" — DSP stretch to the cover box; CPU center-crops.
(The vendor SCALE_AND_CROP picks its own rounding; on device
it disagreed with the CPU placement by ~21 luma levels.)
Hot loops pay no per-call client setup here anymore: the DSP arm
rides the process-resident client (:func:`accel.shared_dsp_call`).
"""
handle = self.handle
if handle is None or handle.closed:
return None
if self.format not in _DSP_RESIZE_FORMATS:
return None
yuv = self.format in _YUV_FORMATS
sw, sh = self.width, self.height
# (rw, rh, ox, oy) from the shared geometry helper — must match
# _resize_yuv/_resize_packed placement, or the two paths would
# disagree on the same frame (and on its transform metadata)
rw, rh, ox, oy = _resize_geometry(sw, sh, dw, dh, mode, yuv)
from .accel import ( # lazy: accel imports frame
HardwareUnavailable,
RoutePolicy,
get_default_router,
shared_dsp_call,
)
from .dsp import DspError # lazy: dsp imports media
router = get_default_router()
try:
wants_hw = router.route("resize_nv12").backend == "hardware"
except KeyError:
wants_hw = True # unregistered op: keep the historical attempt
if not wants_hw:
return None
try:
content = shared_dsp_call("resize_hw", self, rw, rh, scaling="stretch")
except DspError as exc:
logger.debug("DSP resize fast path unavailable (%s); CPU path", exc)
if router.policy is RoutePolicy.HARDWARE_ONLY:
raise HardwareUnavailable(
f"resize: DSP fast path failed ({exc}) and the policy "
"is hardware-only — refusing the silent CPU fallback"
) from exc
# PREFER_HARDWARE degrades, but honestly: the router's
# health()/counters must see the fast path it did not take
router.note_degradation(
"resize_nv12", f"DSP resize fast path unavailable: {exc}"
)
return None
if mode == "stretch":
return content
if not yuv:
if mode == "crop":
return np.ascontiguousarray(content[oy : oy + dh, ox : ox + dw])
canvas = (
np.full((dh, dw), pad, dtype=np.uint8)
if content.ndim == 2
else np.full((dh, dw, content.shape[2]), [pad] * content.shape[2], dtype=np.uint8)
)
canvas[oy : oy + rh, ox : ox + rw] = content
return canvas
if mode == "crop":
return np.ascontiguousarray(
np.vstack(
[
content[:rh][oy : oy + dh, ox : ox + dw],
content[rh:][oy // 2 : oy // 2 + dh // 2, ox : ox + dw],
]
)
)
canvas_y = np.full((dh, dw), pad, dtype=np.uint8)
canvas_uv = np.full((dh // 2, dw), 128, dtype=np.uint8)
canvas_y[oy : oy + rh, ox : ox + rw] = content[:rh]
canvas_uv[oy // 2 : oy // 2 + rh // 2, ox : ox + rw] = content[rh:]
return np.vstack([canvas_y, canvas_uv])
def _resize_packed(self, dw: int, dh: int, mode: str, pad: int) -> np.ndarray:
src = self.image
sw, sh = self.width, self.height
if mode == "stretch":
return _resize_array(src, dw, dh)
rw, rh, ox, oy = _resize_geometry(sw, sh, dw, dh, mode, yuv=False)
if mode == "letterbox":
content = _resize_array(src, rw, rh)
if src.ndim == 2:
canvas = np.full((dh, dw), pad, dtype=np.uint8)
else:
fill = [pad] * src.shape[2]
if self.format in ("RGBA", "BGRA"):
fill[-1] = 255
canvas = np.full((dh, dw, src.shape[2]), fill, dtype=np.uint8)
canvas[oy : oy + rh, ox : ox + rw] = content
return canvas
# mode == "crop": scale to cover, then center-crop
tmp = _resize_array(src, rw, rh)
return np.ascontiguousarray(tmp[oy : oy + dh, ox : ox + dw])
def _resize_yuv(self, dw: int, dh: int, mode: str, pad: int) -> np.ndarray:
sw, sh = self.width, self.height
y_plane = self.image[:sh]
uv_plane = np.ascontiguousarray(self.image[sh:])
if uv_plane.shape[0] * 2 != sh or uv_plane.shape[1] != sw:
raise ValueError(
f"{self.format} buffer shape {self.image.shape} does not match {sw}x{sh} frame"
)
def uv_resize(uv: np.ndarray, w: int, h: int) -> np.ndarray:
# Packed interleaved chroma: resize as (h/2, w/2, 2) image so
# U and V stay on separate channels, then flatten back.
src_h, src_w = uv.shape
paired = uv.reshape(src_h, src_w // 2, 2)
out = _resize_array(paired, w // 2, h // 2)
return out.reshape(h // 2, w)
if mode == "stretch":
return np.vstack(
[
_resize_array(y_plane, dw, dh),
uv_resize(uv_plane, dw, dh),
]
)
rw, rh, ox, oy = _resize_geometry(sw, sh, dw, dh, mode, yuv=True)
if mode == "letterbox":
content_y = _resize_array(y_plane, rw, rh)
content_uv = uv_resize(uv_plane, rw, rh)
canvas_y = np.full((dh, dw), pad, dtype=np.uint8)
canvas_uv = np.full((dh // 2, dw), 128, dtype=np.uint8)
canvas_y[oy : oy + rh, ox : ox + rw] = content_y
canvas_uv[oy // 2 : oy // 2 + rh // 2, ox : ox + rw] = content_uv
return np.vstack([canvas_y, canvas_uv])
# mode == "crop": scale to cover, then center-crop both planes
tmp_y = _resize_array(y_plane, rw, rh)
tmp_uv = uv_resize(uv_plane, rw, rh)
return np.vstack(
[
tmp_y[oy : oy + dh, ox : ox + dw],
tmp_uv[oy // 2 : oy // 2 + dh // 2, ox : ox + dw],
]
)
[docs]
def to_jpeg_bytes(self, quality: int = 85) -> bytes:
"""Encode the frame as JPEG bytes, hardware-first for keep-fd frames.
Frames with a live dma-buf handle ride the accel router's
zero-copy EncodeImage leg (the daemon imports the buffer — no
RGB materialization, no read-back) and degrade to the CPU
cv2/Pillow encode when the daemon cannot serve them. In-memory
frames stay on the CPU encode outright: the daemon encoder is
N-threaded libjpeg behind an RPC, so its win is the zero-copy
import, not raw speed — tight loops (e.g. MjpegStream.push_frame)
must not pay pool-alloc + copy + RPC per frame (S-3 record).
"""
handle = self.handle
if handle is None or handle.closed:
return _encode_jpeg(self.to_rgb(), quality)
from .accel import get_default_router # lazy: accel imports frame
return get_default_router().run("encode_jpeg", self, quality)
[docs]
def save(self, path: str) -> None:
if path.lower().endswith((".jpg", ".jpeg")):
with open(path, "wb") as fh:
fh.write(_encode_jpeg(self.to_rgb()))
return
try:
import cv2
rgb = self.to_rgb()
bgr = rgb[:, :, ::-1]
cv2.imwrite(path, bgr)
except ImportError:
from PIL import Image
rgb = self.to_rgb()
Image.fromarray(rgb).save(path)
[docs]
@dataclass
class StreamInfo:
stream_id: str
width: int
height: int
format: str
fps: float
buffer_count: int
# DMA_BUF_IOCTL_SYNC (linux/dma-buf.h): _IOW('b', 0, u64) on 64-bit.
_DMA_BUF_IOCTL_SYNC = 0x40086200
_DMA_BUF_SYNC_READ = 1 << 0
_DMA_BUF_SYNC_WRITE = 2 << 0
_DMA_BUF_SYNC_START = 1 << 2
_DMA_BUF_SYNC_END = 2 << 2
def _dma_buf_sync(fd: int, flags: int) -> None:
"""Fence CPU access to a dma-buf (HAL-3 discipline).
Device-written dma-bufs must be fenced READ|START before a CPU read
and READ|END after it, or the read can serve stale cache lines. On
non-dma-buf fds (memfd in tests, plain anon memory) the ioctl raises
ENOTTY — there is nothing to fence, so OSError is swallowed.
"""
try:
fcntl.ioctl(fd, _DMA_BUF_IOCTL_SYNC, struct.pack("=Q", flags))
except OSError:
pass
def _decode_raw(raw: np.ndarray, w: int, h: int, fmt: str) -> np.ndarray:
"""Reshape a flat frame buffer into its per-format image layout."""
if fmt in ("NV12", "NV21"):
return raw.reshape(h * 3 // 2, w)
elif fmt in ("RGB", "BGR"):
return raw.reshape(h, w, 3)
elif fmt in ("RGBA", "BGRA"):
return raw.reshape(h, w, 4)
elif fmt == "GRAY8":
return raw.reshape(h, w)
elif fmt == "YUYV":
return raw.reshape(h, w, 2)
return raw.reshape(h, w, 3)
def _materialize_handle(handle: FrameHandle) -> np.ndarray:
"""Map a retained frame's planes and return them as one numpy copy."""
planes = []
for i, fd in enumerate(handle.fds):
size = handle.plane_sizes[i] if i < len(handle.plane_sizes) else 0
_dma_buf_sync(fd, _DMA_BUF_SYNC_READ | _DMA_BUF_SYNC_START)
actual_size = os.fstat(fd).st_size
buf = mmap.mmap(fd, actual_size, access=mmap.ACCESS_READ)
try:
plane = np.frombuffer(buf, dtype=np.uint8)[:size].copy()
finally:
buf.close()
_dma_buf_sync(fd, _DMA_BUF_SYNC_READ | _DMA_BUF_SYNC_END)
planes.append(plane)
return np.concatenate(planes) if len(planes) > 1 else planes[0]