MarilleTielens
By “tiling,” I mean splitting the 4000 × 3000 image into four overlapping regions. Each region is passed to a separate AprilTag node, allowing the four smaller images to be processed concurrently on the OAK4-D. The overlap helps detect tags that fall across tile boundaries. You can see an example here.
With one AprilTag node processing the complete 12 MP image, I measured approximately 30 FPS. With four tiled AprilTag nodes, the on-device benchmark reported approximately 60 FPS for each tile. The test ran as a standalone OAK app and sent only the decoded detections to the host, no image frames were transferred.
Here is also the testing script with tiling:
#!/usr/bin/env python3
import json
import time
import depthai as dai
FULL_WIDTH = 4000
FULL_HEIGHT = 3000
FULL_RES = (FULL_WIDTH, FULL_HEIGHT)
CAMERA_FPS = 60.0
OVERLAP_X = 100
OVERLAP_Y = 80
REPORT_EVERY_N_MESSAGES = 30
HOST_QUEUE_SIZE = 4
# Tile coordinates: x0, y0, x1, y1
TILES = [
# Top-left
(
0,
0,
FULL_WIDTH // 2 + OVERLAP_X,
FULL_HEIGHT // 2 + OVERLAP_Y,
),
# Top-right
(
FULL_WIDTH // 2 - OVERLAP_X,
0,
FULL_WIDTH,
FULL_HEIGHT // 2 + OVERLAP_Y,
),
# Bottom-left
(
0,
FULL_HEIGHT // 2 - OVERLAP_Y,
FULL_WIDTH // 2 + OVERLAP_X,
FULL_HEIGHT,
),
# Bottom-right
(
FULL_WIDTH // 2 - OVERLAP_X,
FULL_HEIGHT // 2 - OVERLAP_Y,
FULL_WIDTH,
FULL_HEIGHT,
),
]
def remap_point(point: dai.Point2f, offset_x: int, offset_y: int) -> dict:
"""Convert tile-local coordinates to full-resolution coordinates."""
return {
"x": float(point.x + offset_x),
"y": float(point.y + offset_y),
}
def serialize_tag(tag, tile_index: int, offset_x: int, offset_y: int) -> dict:
"""Convert one decoded tag into a JSON-serializable dictionary."""
top_left = remap_point(tag.topLeft, offset_x, offset_y)
top_right = remap_point(tag.topRight, offset_x, offset_y)
bottom_right = remap_point(tag.bottomRight, offset_x, offset_y)
bottom_left = remap_point(tag.bottomLeft, offset_x, offset_y)
return {
"tile": tile_index,
"id": int(tag.id),
"hamming": int(tag.hamming),
"decisionMargin": float(tag.decisionMargin),
"center": {
"x": (top_left["x"] + bottom_right["x"]) / 2.0,
"y": (top_left["y"] + bottom_right["y"]) / 2.0,
},
"corners": {
"topLeft": top_left,
"topRight": top_right,
"bottomRight": bottom_right,
"bottomLeft": bottom_left,
},
}
with dai.Pipeline() as pipeline:
camera = pipeline.create(dai.node.Camera).build(
dai.CameraBoardSocket.CAM_A
)
# Request 4000 x 3000 NV12 at 60 FPS.
camera_capability = dai.ImgFrameCapability()
camera_capability.size.fixed(FULL_RES)
camera_capability.fps.fixed(CAMERA_FPS)
camera_capability.type = dai.ImgFrame.Type.NV12
camera_capability.resizeMode = dai.ImgResizeMode.CROP
full_output = camera.requestOutput(
camera_capability,
True,
)
# Each entry contains the host queue and the tile offset required to map
# detections back into the full 4000 x 3000 coordinate system.
detection_streams = []
for tile_index, tile in enumerate(TILES):
x0, y0, x1, y1 = tile
tile_width = x1 - x0
tile_height = y1 - y0
tile_manip = pipeline.create(dai.node.ImageManip)
tile_manip.initialConfig.addCrop(
x0,
y0,
tile_width,
tile_height,
)
tile_manip.setMaxOutputFrameSize(
tile_width * tile_height * 3
)
full_output.link(tile_manip.inputImage)
april_tag = pipeline.create(dai.node.AprilTag)
april_tag.initialConfig.setFamily(
dai.AprilTagConfig.Family.TAG_36H11
)
april_tag.initialConfig.quadSigma = 0.0
april_tag.initialConfig.refineEdges = True
april_tag.initialConfig.decodeSharpening = 0.25
april_tag.initialConfig.maxHammingDistance = 1
tile_manip.out.link(april_tag.inputImage)
# Benchmark each tile entirely on-device.
benchmark = pipeline.create(dai.node.BenchmarkIn)
benchmark.setRunOnHost(False)
benchmark.sendReportEveryNMessages(
REPORT_EVERY_N_MESSAGES
)
benchmark.logReportsAsWarnings(True)
april_tag.out.link(benchmark.input)
# Also forward the decoded AprilTags messages to the host. This queue
# is non-blocking, so a slow consumer will not stall the pipeline.
detection_queue = april_tag.out.createOutputQueue(
maxSize=HOST_QUEUE_SIZE,
blocking=False,
)
detection_streams.append(
{
"tile": tile_index,
"offset_x": x0,
"offset_y": y0,
"queue": detection_queue,
}
)
print(
f"Tile {tile_index}: "
f"x={x0}:{x1}, "
f"y={y0}:{y1}, "
f"size={tile_width}x{tile_height}",
flush=True,
)
pipeline.start()
print("Tiled AprilTag pipeline started", flush=True)
print(
f"Camera request: "
f"{FULL_WIDTH}x{FULL_HEIGHT} @ {CAMERA_FPS} FPS",
flush=True,
)
print(f"Tiles: {len(TILES)}", flush=True)
print(
"Decoded detections are forwarded to host queues; "
"image frames are not forwarded.",
flush=True,
)
try:
while pipeline.isRunning():
for stream in detection_streams:
while True:
message = stream["queue"].tryGet()
if message is None:
break
assert isinstance(message, dai.AprilTags)
tags = [
serialize_tag(
tag,
stream["tile"],
stream["offset_x"],
stream["offset_y"],
)
for tag in message.aprilTags
]
# Avoid flooding stdout with empty detection messages.
if tags:
print(
json.dumps(
{
"type": "apriltags",
"tile": stream["tile"],
"sequence": message.getSequenceNum(),
"timestampDevice": (
message
.getTimestampDevice()
.total_seconds()
),
"tags": tags,
},
separators=(",", ":"),
),
flush=True,
)
# Prevent a non-blocking polling loop from consuming an entire
# OAK4 host CPU core.
time.sleep(0.001)
except KeyboardInterrupt:
print("Stopping pipeline.", flush=True)