Merge branch 'isp_dma_one_frame_v6.0' into 'release/v6.0'

feat(isp): support isp dma input and add example (v6.0)

See merge request espressif/esp-idf!50755
This commit is contained in:
morris
2026-07-16 17:17:25 +08:00
20 changed files with 1358 additions and 19 deletions
@@ -228,6 +228,14 @@ examples/peripherals/i3c/i3c_i2c_basic:
- *common_components
- esp_driver_i3c
examples/peripherals/isp/dma_input:
disable:
- if: SOC_ISP_SUPPORTED != 1
depends_components:
- esp_driver_dma
- esp_driver_isp
- soc
examples/peripherals/isp/multi_pipelines:
disable:
- if: SOC_MIPI_CSI_SUPPORTED != 1
@@ -0,0 +1,8 @@
# The following lines of boilerplate have to be in your project's CMakeLists
# in this exact order for cmake to work correctly
cmake_minimum_required(VERSION 3.22)
include($ENV{IDF_PATH}/tools/cmake/project.cmake)
# "Trim" the build. Include the minimal set of components, main, and anything it depends on.
idf_build_set_property(MINIMAL_BUILD ON)
project(isp_dma_input)
@@ -0,0 +1,49 @@
| Supported Targets | ESP32-P4 |
| ----------------- | -------- |
# ISP DMA Input Visual Test Example
## Overview
This example embeds a 240 x 280 RAW8 Bayer image of a real scene in flash, copies it into a DMA-capable PSRAM input buffer, feeds it into the ISP through DW-GDMA, and prints the RGB888 output as base64. The pytest script decodes the output into a PPM image and compares it with the checked-in golden image.
The data flow is:
1. The embedded BGGR RAW8 image is copied from flash into the ISP DMA input buffer.
2. The image is transferred into the ISP via `DW-GDMA → ISP DMA input`.
3. The ISP processes the data (demosaic, color adjustment) and outputs RGB888 (BGR24 byte layout).
4. The RGB888 frame is base64-encoded and printed with machine-parseable markers.
5. pytest decodes the payload, swaps BGR→RGB, saves one PPM file per frame, and compares it with the golden image.
## Hardware Required
- An ESP32-P4 devkit with PSRAM (this example allocates the ISP DMA input/output buffers from PSRAM).
## How to Use
Run the test locally. It builds and flashes the example, captures the serial output, and saves one PPM artifact per frame:
```
cd examples/peripherals/isp/dma_input
pytest pytest_isp_dma_input.py --target esp32p4 --port PORT
```
The pytest log directory (`dut.logdir`) contains `isp_dma_input_frame00.ppm`, the decoded RGB888 ISP output from the current hardware run. The repository includes [golden/golden.ppm](golden/golden.ppm), the checked-in reference image. The test compares the decoded RGB888 pixels with this reference image, making image-quality regressions visible in review and detectable in CI.
## Example Output
The example processes one embedded 240 x 280 BGGR RAW8 frame.
```text
Feeding 1 frames through ISP DMA input...
IMAGE_META frame=0 width=240 height=280 format=BGR24 encoding=base64
IMAGE_BASE64_BEGIN
IMAGE_BASE64 ...
IMAGE_BASE64_END
Frame 0 done
ISP DMA visual demo done.
```
## Reference
- [ESP-IDF: Image Signal Processor](https://docs.espressif.com/projects/esp-idf/en/latest/esp32p4/api-reference/peripherals/isp.html)
Binary file not shown.
@@ -0,0 +1,7 @@
idf_component_register(SRCS "isp_dma_main.c"
PRIV_REQUIRES esp_driver_isp esp_mm esp_psram mbedtls
INCLUDE_DIRS ".")
target_add_binary_data(${COMPONENT_LIB}
"${CMAKE_CURRENT_LIST_DIR}/assets/sensor_240x280_bggr.raw"
BINARY RENAME_TO "sensor_raw")
File diff suppressed because one or more lines are too long
@@ -0,0 +1,128 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "mbedtls/base64.h"
#include "esp_check.h"
#include "esp_cache.h"
#include "esp_heap_caps.h"
#include "driver/isp_dma.h"
#include "driver/isp_core.h"
#include "driver/isp_color.h"
#define EXAMPLE_WIDTH 240
#define EXAMPLE_HEIGHT 280
#define EXAMPLE_BASE64_CHUNK_LEN 384
#define EXAMPLE_BASE64_DELAY_MS 10
#define EXAMPLE_DMA_ALIGN 64
#define EXAMPLE_FRAME_COUNT 1
extern const uint8_t sensor_raw_start[] asm("_binary_sensor_raw_start");
extern const uint8_t sensor_raw_end[] asm("_binary_sensor_raw_end");
static void *s_alloc_dma_buffer(size_t size)
{
return heap_caps_aligned_calloc(EXAMPLE_DMA_ALIGN, 1, size,
MALLOC_CAP_SPIRAM | MALLOC_CAP_DMA | MALLOC_CAP_8BIT);
}
static void s_configure_neutral_color(isp_proc_handle_t isp_proc)
{
esp_isp_color_config_t color_cfg = {
.color_contrast = { .integer = 1, .decimal = 0 },
.color_saturation = { .integer = 1, .decimal = 0 },
.color_hue = 0,
.color_brightness = 0,
};
ESP_ERROR_CHECK(esp_isp_color_configure(isp_proc, &color_cfg));
ESP_ERROR_CHECK(esp_isp_color_enable(isp_proc));
}
static void s_print_base64_payload(const unsigned char *encoded, size_t encoded_len)
{
printf("IMAGE_BASE64_BEGIN\n");
fflush(stdout);
for (size_t offset = 0; offset < encoded_len; offset += EXAMPLE_BASE64_CHUNK_LEN) {
size_t chunk_len = encoded_len - offset;
if (chunk_len > EXAMPLE_BASE64_CHUNK_LEN) {
chunk_len = EXAMPLE_BASE64_CHUNK_LEN;
}
printf("IMAGE_BASE64 %.*s\n", (int)chunk_len, (const char *)&encoded[offset]);
fflush(stdout);
vTaskDelay(pdMS_TO_TICKS(EXAMPLE_BASE64_DELAY_MS));
}
printf("IMAGE_BASE64_END\n");
fflush(stdout);
}
void app_main(void)
{
const uint32_t h_res = EXAMPLE_WIDTH;
const uint32_t v_res = EXAMPLE_HEIGHT;
const size_t in_size = (size_t)h_res * v_res; // RAW8: 1 byte/pixel
const size_t out_size = (size_t)h_res * v_res * 3; // RGB888: 3 bytes/pixel
isp_proc_handle_t isp_proc = NULL;
esp_isp_processor_cfg_t isp_cfg = {
.clk_hz = 240 * 1000 * 1000,
.input_data_source = ISP_INPUT_DATA_SOURCE_DWGDMA,
.input_data_color_type = ISP_COLOR_RAW8,
.output_data_color_type = ISP_COLOR_RGB888,
.bayer_order = COLOR_RAW_ELEMENT_ORDER_BGGR,
.has_line_start_packet = false,
.has_line_end_packet = false,
.h_res = h_res,
.v_res = v_res,
.dma_burst_size = 8,
};
ESP_ERROR_CHECK(esp_isp_new_processor(&isp_cfg, &isp_proc));
ESP_ERROR_CHECK(esp_isp_enable(isp_proc));
s_configure_neutral_color(isp_proc);
uint8_t *isp_in_buf = s_alloc_dma_buffer(in_size);
uint8_t *isp_out_buf = s_alloc_dma_buffer(out_size);
assert(isp_in_buf && isp_out_buf);
size_t embedded_raw_size = sensor_raw_end - sensor_raw_start;
assert(embedded_raw_size == in_size);
memcpy(isp_in_buf, sensor_raw_start, embedded_raw_size);
size_t encoded_len = 0;
int ret = mbedtls_base64_encode(NULL, 0, &encoded_len, isp_out_buf, out_size);
ESP_ERROR_CHECK((ret == MBEDTLS_ERR_BASE64_BUFFER_TOO_SMALL) ? ESP_OK : ESP_FAIL);
unsigned char *encoded = calloc(encoded_len + 1, 1);
assert(encoded);
printf("Feeding %d frames through ISP DMA input...\n", EXAMPLE_FRAME_COUNT);
for (int frame = 0; frame < EXAMPLE_FRAME_COUNT; frame++) {
ESP_ERROR_CHECK(esp_cache_msync(isp_in_buf, in_size, ESP_CACHE_MSYNC_FLAG_DIR_C2M));
ESP_ERROR_CHECK(esp_isp_dma_process_frame(isp_proc, isp_out_buf, isp_in_buf, 1000));
ESP_ERROR_CHECK(esp_cache_msync(isp_out_buf, out_size, ESP_CACHE_MSYNC_FLAG_DIR_M2C));
size_t out_len = 0;
ESP_ERROR_CHECK(mbedtls_base64_encode(encoded, encoded_len + 1, &out_len, isp_out_buf, out_size) == 0 ? ESP_OK : ESP_FAIL);
printf("IMAGE_META frame=%d width=%u height=%u format=BGR24 encoding=base64\n",
frame, (unsigned)h_res, (unsigned)v_res);
s_print_base64_payload(encoded, out_len);
printf("Frame %d done\n", frame);
}
printf("ISP DMA visual demo done.\n");
free(encoded);
ESP_ERROR_CHECK(esp_isp_color_disable(isp_proc));
ESP_ERROR_CHECK(esp_isp_disable(isp_proc));
ESP_ERROR_CHECK(esp_isp_del_processor(isp_proc));
heap_caps_free(isp_in_buf);
heap_caps_free(isp_out_buf);
}
@@ -0,0 +1,181 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: CC0-1.0
import base64
import logging
import re
from dataclasses import dataclass
from pathlib import Path
import pytest
from pytest_embedded import Dut
from pytest_embedded_idf.utils import idf_parametrize
from pytest_embedded_idf.utils import soc_filtered_targets
IMAGE_META_PATTERN = (
r'IMAGE_META frame=(?P<frame>\d+) width=(?P<width>\d+) height=(?P<height>\d+) '
r'format=(?P<format>\w+) encoding=(?P<encoding>\w+)'
)
IMAGE_META_RE = re.compile(IMAGE_META_PATTERN)
IMAGE_CHUNK_PATTERN = (
r'IMAGE_BASE64 (?P<payload>[A-Za-z0-9+/=]+?)'
r'(?=\r?\n|IMAGE_BASE64|IMAGE_BASE64_END|Frame )'
)
IMAGE_OUTPUT_TEMPLATE = 'isp_dma_input_frame{frame:02d}.ppm'
REFERENCE_IMAGE_PATH = Path(__file__).parent / 'golden' / 'golden.ppm'
EXPECTED_PIXEL_FORMAT = 'BGR24'
EXPECTED_ENCODING = 'base64'
RGB888_BYTES_PER_PIXEL = 3
PPM_MAGIC = b'P6'
PPM_MAX_VALUE = b'255'
@dataclass(frozen=True)
class ImageMetadata:
frame: int
width: int
height: int
pixel_format: str
encoding: str
@dataclass(frozen=True)
class RgbImage:
width: int
height: int
pixels_rgb888: bytes
def __post_init__(self) -> None:
expected_size = self.width * self.height * RGB888_BYTES_PER_PIXEL
if len(self.pixels_rgb888) != expected_size:
raise ValueError(f'Expected {expected_size} RGB bytes, got {len(self.pixels_rgb888)}')
def parse_image_metadata(meta_line: str) -> ImageMetadata:
match = IMAGE_META_RE.fullmatch(meta_line)
if not match:
raise ValueError(f'Invalid image metadata line: {meta_line}')
return ImageMetadata(
frame=int(match.group('frame')),
width=int(match.group('width')),
height=int(match.group('height')),
pixel_format=match.group('format'),
encoding=match.group('encoding'),
)
def collect_base64_payload(dut: Dut) -> list[str]:
payload_lines: list[str] = []
while True:
match = dut.expect(rf'IMAGE_BASE64_END|{IMAGE_CHUNK_PATTERN}', timeout=60)
if match.group(0).decode('utf-8') == 'IMAGE_BASE64_END':
return payload_lines
payload_lines.append(match.group('payload').decode('utf-8'))
def _bgr24_to_rgb888(raw_bytes: bytes) -> bytes:
rgb_bytes = bytearray(len(raw_bytes))
for offset in range(0, len(raw_bytes), 3):
blue, green, red = raw_bytes[offset : offset + 3]
rgb_bytes[offset : offset + 3] = (red, green, blue)
return bytes(rgb_bytes)
def _encode_ppm(image: RgbImage) -> bytes:
header = b'%s\n%d %d\n%s\n' % (PPM_MAGIC, image.width, image.height, PPM_MAX_VALUE)
return header + image.pixels_rgb888
def decode_bgr24_base64_image(metadata: ImageMetadata, payload_lines: list[str]) -> RgbImage:
if metadata.pixel_format != EXPECTED_PIXEL_FORMAT:
raise ValueError(f'Unsupported pixel format: {metadata.pixel_format}')
if metadata.encoding != EXPECTED_ENCODING:
raise ValueError(f'Unsupported payload encoding: {metadata.encoding}')
raw_bytes = base64.b64decode(''.join(payload_lines), validate=True)
expected_size = metadata.width * metadata.height * RGB888_BYTES_PER_PIXEL
if len(raw_bytes) != expected_size:
raise ValueError(f'Expected {expected_size} decoded bytes, got {len(raw_bytes)}')
return RgbImage(width=metadata.width, height=metadata.height, pixels_rgb888=_bgr24_to_rgb888(raw_bytes))
def save_ppm_artifact(image: RgbImage, output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
try:
output_path.write_bytes(_encode_ppm(image))
except OSError:
logging.exception('Failed to save ISP DMA artifact to %s', output_path)
return
logging.info('Saved ISP DMA artifact to %s', output_path)
def load_ppm_image(image_path: Path) -> RgbImage:
ppm_data = image_path.read_bytes()
try:
magic, dimensions, max_value, pixels_rgb888 = ppm_data.split(b'\n', 3)
width, height = (int(value) for value in dimensions.split())
except ValueError as error:
raise ValueError(f'Invalid PPM image: {image_path}') from error
if magic != PPM_MAGIC or max_value != PPM_MAX_VALUE:
raise ValueError(f'Unsupported PPM image: {image_path}')
return RgbImage(width=width, height=height, pixels_rgb888=pixels_rgb888)
def assert_image_matches_reference(result_image: RgbImage, reference_image: RgbImage) -> None:
assert (result_image.width, result_image.height) == (reference_image.width, reference_image.height), (
f'ISP output dimensions do not match reference image: '
f'{result_image.width}x{result_image.height} != {reference_image.width}x{reference_image.height}'
)
if result_image.pixels_rgb888 == reference_image.pixels_rgb888:
return
mismatch_offset = next(
offset
for offset, (actual, expected) in enumerate(zip(result_image.pixels_rgb888, reference_image.pixels_rgb888))
if actual != expected
)
pixel_index, channel = divmod(mismatch_offset, RGB888_BYTES_PER_PIXEL)
raise AssertionError(
f'ISP output does not match reference image at pixel {pixel_index}, channel {channel}: '
f'{result_image.pixels_rgb888[mismatch_offset]} != {reference_image.pixels_rgb888[mismatch_offset]}'
)
@pytest.mark.generic
@idf_parametrize('target', soc_filtered_targets('SOC_ISP_SUPPORTED == 1'), indirect=['target'])
def test_isp_dma_input_example(dut: Dut) -> None:
reference_image = load_ppm_image(REFERENCE_IMAGE_PATH)
frame_count_match = dut.expect(r'Feeding (?P<frame_count>\d+) frames through ISP DMA input...')
expected_frame_count = int(frame_count_match.group('frame_count').decode('utf-8'))
logging.info('Expecting %d ISP DMA frame(s)', expected_frame_count)
for expected_frame in range(expected_frame_count):
metadata = parse_image_metadata(dut.expect(IMAGE_META_PATTERN).group(0).decode('utf-8'))
assert metadata.frame == expected_frame, f'Out-of-order frame: expected {expected_frame}, got {metadata.frame}'
logging.info(
'Received frame metadata: frame=%d size=%dx%d format=%s',
metadata.frame,
metadata.width,
metadata.height,
metadata.pixel_format,
)
dut.expect_exact('IMAGE_BASE64_BEGIN')
logging.info('Receiving base64 image payload for frame %d', metadata.frame)
payload_lines = collect_base64_payload(dut)
logging.info('Received %d base64 chunk(s) for frame %d', len(payload_lines), metadata.frame)
result_image = decode_bgr24_base64_image(metadata, payload_lines)
logging.info('Decoded frame %d to RGB888 image', metadata.frame)
output_path = Path(dut.logdir) / IMAGE_OUTPUT_TEMPLATE.format(frame=metadata.frame)
save_ppm_artifact(result_image, output_path)
assert_image_matches_reference(result_image, reference_image)
dut.expect_exact(f'Frame {expected_frame} done')
dut.expect_exact('ISP DMA visual demo done.')
@@ -0,0 +1 @@
CONFIG_SPIRAM=y