mirror of
https://github.com/esphome/esphome.git
synced 2026-08-22 22:26:21 +00:00
[api] Create the camera image reader lazily (#18421)
This commit is contained in:
@@ -160,11 +160,6 @@ APIConnection::APIConnection(std::unique_ptr<socket::Socket> sock, APIServer *pa
|
|||||||
#else
|
#else
|
||||||
#error "No frame helper defined"
|
#error "No frame helper defined"
|
||||||
#endif
|
#endif
|
||||||
#ifdef USE_CAMERA
|
|
||||||
if (camera::Camera::instance() != nullptr) {
|
|
||||||
this->image_reader_ = std::unique_ptr<camera::CameraImageReader>{camera::Camera::instance()->create_image_reader()};
|
|
||||||
}
|
|
||||||
#endif
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void APIConnection::start() {
|
void APIConnection::start() {
|
||||||
@@ -1140,6 +1135,7 @@ void APIConnection::try_send_camera_image_() {
|
|||||||
if (!this->image_reader_)
|
if (!this->image_reader_)
|
||||||
return;
|
return;
|
||||||
|
|
||||||
|
const auto *cam = camera::Camera::instance();
|
||||||
// Send as many chunks as possible without blocking
|
// Send as many chunks as possible without blocking
|
||||||
while (this->image_reader_->available()) {
|
while (this->image_reader_->available()) {
|
||||||
if (!this->helper_->can_write_without_blocking())
|
if (!this->helper_->can_write_without_blocking())
|
||||||
@@ -1149,11 +1145,11 @@ void APIConnection::try_send_camera_image_() {
|
|||||||
bool done = this->image_reader_->available() == to_send;
|
bool done = this->image_reader_->available() == to_send;
|
||||||
|
|
||||||
CameraImageResponse msg;
|
CameraImageResponse msg;
|
||||||
msg.key = camera::Camera::instance()->get_object_id_hash();
|
msg.key = cam->get_object_id_hash();
|
||||||
msg.set_data(this->image_reader_->peek_data_buffer(), to_send);
|
msg.set_data(this->image_reader_->peek_data_buffer(), to_send);
|
||||||
msg.done = done;
|
msg.done = done;
|
||||||
#ifdef USE_DEVICES
|
#ifdef USE_DEVICES
|
||||||
msg.device_id = camera::Camera::instance()->get_device_id();
|
msg.device_id = cam->get_device_id();
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
if (!this->send_message(msg)) {
|
if (!this->send_message(msg)) {
|
||||||
@@ -1169,15 +1165,19 @@ void APIConnection::try_send_camera_image_() {
|
|||||||
void APIConnection::set_camera_state(std::shared_ptr<camera::CameraImage> image) {
|
void APIConnection::set_camera_state(std::shared_ptr<camera::CameraImage> image) {
|
||||||
if (!this->flags_.state_subscription)
|
if (!this->flags_.state_subscription)
|
||||||
return;
|
return;
|
||||||
if (!this->image_reader_)
|
if (this->image_reader_ && this->image_reader_->available())
|
||||||
return;
|
return;
|
||||||
if (this->image_reader_->available())
|
if (!image->was_requested_by(esphome::camera::API_REQUESTER) && !image->was_requested_by(esphome::camera::IDLE))
|
||||||
return;
|
return;
|
||||||
if (image->was_requested_by(esphome::camera::API_REQUESTER) || image->was_requested_by(esphome::camera::IDLE)) {
|
if (!this->image_reader_) {
|
||||||
this->image_reader_->set_image(std::move(image));
|
// Created on the first image this connection will send, so connections
|
||||||
// Try to send immediately to reduce latency
|
// that never receive one never pay for a reader. Only a registered
|
||||||
this->try_send_camera_image_();
|
// camera's listener can reach this, so instance() is non-null here.
|
||||||
|
this->image_reader_ = std::unique_ptr<camera::CameraImageReader>{camera::Camera::instance()->create_image_reader()};
|
||||||
}
|
}
|
||||||
|
this->image_reader_->set_image(std::move(image));
|
||||||
|
// Try to send immediately to reduce latency
|
||||||
|
this->try_send_camera_image_();
|
||||||
}
|
}
|
||||||
uint16_t APIConnection::try_send_camera_info(EntityBase *entity, APIConnection *conn, uint32_t remaining_size) {
|
uint16_t APIConnection::try_send_camera_info(EntityBase *entity, APIConnection *conn, uint32_t remaining_size) {
|
||||||
auto *camera = static_cast<camera::Camera *>(entity);
|
auto *camera = static_cast<camera::Camera *>(entity);
|
||||||
|
|||||||
@@ -103,7 +103,8 @@ struct CameraImageSpec {
|
|||||||
/** Abstract camera base class. Collaborates with API.
|
/** Abstract camera base class. Collaborates with API.
|
||||||
* 1) API server starts and registers as a listener (add_listener)
|
* 1) API server starts and registers as a listener (add_listener)
|
||||||
* to receive new images from the camera.
|
* to receive new images from the camera.
|
||||||
* 2) New API client connects and creates a new image reader (create_image_reader).
|
* 2) API connection creates an image reader (create_image_reader) when it receives
|
||||||
|
* the first image it will send.
|
||||||
* 3) API connection receives protobuf CameraImageRequest and calls request_image.
|
* 3) API connection receives protobuf CameraImageRequest and calls request_image.
|
||||||
* 3.a) API connection receives protobuf CameraImageRequest and calls start_stream.
|
* 3.a) API connection receives protobuf CameraImageRequest and calls start_stream.
|
||||||
* 4) Camera implementation provides JPEG data in the CameraImage and notifies listeners.
|
* 4) Camera implementation provides JPEG data in the CameraImage and notifies listeners.
|
||||||
|
|||||||
@@ -0,0 +1,19 @@
|
|||||||
|
esphome:
|
||||||
|
name: camera-mock-test
|
||||||
|
|
||||||
|
host:
|
||||||
|
api:
|
||||||
|
logger:
|
||||||
|
level: VERBOSE
|
||||||
|
|
||||||
|
external_components:
|
||||||
|
- source:
|
||||||
|
type: local
|
||||||
|
path: EXTERNAL_COMPONENT_PATH
|
||||||
|
|
||||||
|
mock_camera:
|
||||||
|
name: Mock Camera
|
||||||
|
# Larger than MAX_BATCH_PACKET_SIZE (1390) so the image is split across
|
||||||
|
# multiple CameraImageResponse chunks and the client must reassemble.
|
||||||
|
# Must match IMAGE_SIZE in test_camera_mock.py.
|
||||||
|
image_size: 4096
|
||||||
@@ -0,0 +1,28 @@
|
|||||||
|
import esphome.codegen as cg
|
||||||
|
import esphome.config_validation as cv
|
||||||
|
from esphome.const import CONF_ID
|
||||||
|
from esphome.core.entity_helpers import setup_entity
|
||||||
|
from esphome.types import ConfigType
|
||||||
|
|
||||||
|
CODEOWNERS = ["@esphome/tests"]
|
||||||
|
AUTO_LOAD = ["camera"]
|
||||||
|
|
||||||
|
CONF_IMAGE_SIZE = "image_size"
|
||||||
|
|
||||||
|
mock_camera_ns = cg.esphome_ns.namespace("mock_camera")
|
||||||
|
MockCamera = mock_camera_ns.class_("MockCamera", cg.Component, cg.EntityBase)
|
||||||
|
|
||||||
|
CONFIG_SCHEMA = cv.ENTITY_BASE_SCHEMA.extend(
|
||||||
|
{
|
||||||
|
cv.GenerateID(): cv.declare_id(MockCamera),
|
||||||
|
cv.Optional(CONF_IMAGE_SIZE, default=1024): cv.positive_not_null_int,
|
||||||
|
}
|
||||||
|
).extend(cv.COMPONENT_SCHEMA)
|
||||||
|
|
||||||
|
|
||||||
|
async def to_code(config: ConfigType) -> None:
|
||||||
|
cg.add_define("USE_CAMERA")
|
||||||
|
var = cg.new_Pvariable(config[CONF_ID])
|
||||||
|
await setup_entity(var, config, "camera")
|
||||||
|
await cg.register_component(var, config)
|
||||||
|
cg.add(var.set_image_size(config[CONF_IMAGE_SIZE]))
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
#include "mock_camera.h"
|
||||||
|
#include "esphome/core/application.h"
|
||||||
|
#include "esphome/core/log.h"
|
||||||
|
|
||||||
|
namespace esphome::mock_camera {
|
||||||
|
|
||||||
|
static const char *const TAG = "mock_camera";
|
||||||
|
|
||||||
|
void MockCamera::loop() {
|
||||||
|
uint8_t requesters = this->single_requesters_ | this->stream_requesters_;
|
||||||
|
if (requesters == 0)
|
||||||
|
return;
|
||||||
|
uint32_t now = App.get_loop_component_start_time();
|
||||||
|
if (now - this->last_frame_ms_ < FRAME_INTERVAL_MS)
|
||||||
|
return;
|
||||||
|
this->last_frame_ms_ = now;
|
||||||
|
this->single_requesters_ = 0;
|
||||||
|
|
||||||
|
auto image = std::make_shared<MockCameraImage>(this->image_size_, this->frame_counter_, requesters);
|
||||||
|
ESP_LOGV(TAG, "Producing frame %u (%u bytes, requesters 0x%02X)", this->frame_counter_, this->image_size_,
|
||||||
|
requesters);
|
||||||
|
this->frame_counter_++;
|
||||||
|
for (auto *listener : this->listeners_) {
|
||||||
|
listener->on_camera_image(image);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void MockCamera::dump_config() { ESP_LOGCONFIG(TAG, "Mock Camera (%u byte frames)", this->image_size_); }
|
||||||
|
|
||||||
|
} // namespace esphome::mock_camera
|
||||||
@@ -0,0 +1,80 @@
|
|||||||
|
#pragma once
|
||||||
|
|
||||||
|
#include "esphome/components/camera/camera.h"
|
||||||
|
#include "esphome/core/component.h"
|
||||||
|
|
||||||
|
#include <memory>
|
||||||
|
#include <vector>
|
||||||
|
|
||||||
|
namespace esphome::mock_camera {
|
||||||
|
|
||||||
|
/** Deterministic in-memory camera image.
|
||||||
|
* Byte i of frame N is (N + i) & 0xFF so tests can validate
|
||||||
|
* reassembled data from just the first byte.
|
||||||
|
*/
|
||||||
|
class MockCameraImage : public camera::CameraImage {
|
||||||
|
public:
|
||||||
|
MockCameraImage(size_t size, uint8_t frame_counter, uint8_t requesters)
|
||||||
|
: data_(new uint8_t[size]), size_(size), requesters_(requesters) {
|
||||||
|
for (size_t i = 0; i < size; i++) {
|
||||||
|
this->data_[i] = static_cast<uint8_t>(frame_counter + i);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
uint8_t *get_data_buffer() override { return this->data_.get(); }
|
||||||
|
size_t get_data_length() override { return this->size_; }
|
||||||
|
bool was_requested_by(camera::CameraRequester requester) const override {
|
||||||
|
return (this->requesters_ & (1 << requester)) != 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
protected:
|
||||||
|
std::unique_ptr<uint8_t[]> data_;
|
||||||
|
size_t size_;
|
||||||
|
uint8_t requesters_;
|
||||||
|
};
|
||||||
|
|
||||||
|
class MockCameraImageReader : public camera::CameraImageReader {
|
||||||
|
public:
|
||||||
|
void set_image(std::shared_ptr<camera::CameraImage> image) override {
|
||||||
|
this->image_ = std::move(image);
|
||||||
|
this->offset_ = 0;
|
||||||
|
}
|
||||||
|
size_t available() const override { return this->image_ ? this->image_->get_data_length() - this->offset_ : 0; }
|
||||||
|
uint8_t *peek_data_buffer() override { return this->image_->get_data_buffer() + this->offset_; }
|
||||||
|
void consume_data(size_t consumed) override { this->offset_ += consumed; }
|
||||||
|
void return_image() override {
|
||||||
|
this->image_.reset();
|
||||||
|
this->offset_ = 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
protected:
|
||||||
|
std::shared_ptr<camera::CameraImage> image_;
|
||||||
|
size_t offset_{0};
|
||||||
|
};
|
||||||
|
|
||||||
|
/** Virtual camera producing deterministic frames on request or stream. */
|
||||||
|
class MockCamera : public camera::Camera {
|
||||||
|
public:
|
||||||
|
void loop() override;
|
||||||
|
void dump_config() override;
|
||||||
|
|
||||||
|
void add_listener(camera::CameraListener *listener) override { this->listeners_.push_back(listener); }
|
||||||
|
camera::CameraImageReader *create_image_reader() override { return new MockCameraImageReader(); }
|
||||||
|
void request_image(camera::CameraRequester requester) override { this->single_requesters_ |= (1 << requester); }
|
||||||
|
void start_stream(camera::CameraRequester requester) override { this->stream_requesters_ |= (1 << requester); }
|
||||||
|
void stop_stream(camera::CameraRequester requester) override { this->stream_requesters_ &= ~(1 << requester); }
|
||||||
|
|
||||||
|
void set_image_size(uint32_t size) { this->image_size_ = size; }
|
||||||
|
|
||||||
|
protected:
|
||||||
|
static constexpr uint32_t FRAME_INTERVAL_MS = 50;
|
||||||
|
|
||||||
|
// Members ordered largest to smallest to minimize padding
|
||||||
|
std::vector<camera::CameraListener *> listeners_;
|
||||||
|
uint32_t image_size_{1024};
|
||||||
|
uint32_t last_frame_ms_{0};
|
||||||
|
uint8_t frame_counter_{0};
|
||||||
|
uint8_t single_requesters_{0};
|
||||||
|
uint8_t stream_requesters_{0};
|
||||||
|
};
|
||||||
|
|
||||||
|
} // namespace esphome::mock_camera
|
||||||
@@ -0,0 +1,73 @@
|
|||||||
|
"""Integration test for the camera API flow using a mock camera platform."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
from aioesphomeapi import CameraInfo, CameraState, EntityState
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from .state_utils import require_entity
|
||||||
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||||
|
|
||||||
|
# Must match image_size in fixtures/camera_mock.yaml
|
||||||
|
IMAGE_SIZE = 4096
|
||||||
|
STREAM_FRAMES = 3
|
||||||
|
|
||||||
|
|
||||||
|
def _verify_frame(data: bytes) -> int:
|
||||||
|
"""Verify the deterministic frame pattern and return the frame counter."""
|
||||||
|
assert len(data) == IMAGE_SIZE, f"expected {IMAGE_SIZE} bytes, got {len(data)}"
|
||||||
|
counter = data[0]
|
||||||
|
assert data == bytes((counter + i) & 0xFF for i in range(IMAGE_SIZE)), (
|
||||||
|
"frame pattern mismatch"
|
||||||
|
)
|
||||||
|
return counter
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_camera_mock(
|
||||||
|
yaml_config: str,
|
||||||
|
run_compiled: RunCompiledFunction,
|
||||||
|
api_client_connected: APIClientConnectedFactory,
|
||||||
|
) -> None:
|
||||||
|
"""Single-image and stream requests deliver reassembled deterministic frames."""
|
||||||
|
async with run_compiled(yaml_config), api_client_connected() as client:
|
||||||
|
entities, _ = await client.list_entities_services()
|
||||||
|
camera = require_entity(entities, "mock_camera", CameraInfo)
|
||||||
|
|
||||||
|
loop = asyncio.get_running_loop()
|
||||||
|
images: list[bytes] = []
|
||||||
|
single_image: asyncio.Future[None] = loop.create_future()
|
||||||
|
stream_done: asyncio.Future[None] = loop.create_future()
|
||||||
|
|
||||||
|
def on_state(state: EntityState) -> None:
|
||||||
|
if not (isinstance(state, CameraState) and state.key == camera.key):
|
||||||
|
return
|
||||||
|
images.append(bytes(state.data))
|
||||||
|
if not single_image.done():
|
||||||
|
single_image.set_result(None)
|
||||||
|
elif len(images) >= STREAM_FRAMES and not stream_done.done():
|
||||||
|
stream_done.set_result(None)
|
||||||
|
|
||||||
|
client.subscribe_states(on_state)
|
||||||
|
|
||||||
|
# Single image request: one complete frame arrives, reassembled
|
||||||
|
# from multiple chunks (4096 > 1390 byte packets)
|
||||||
|
client.request_single_image()
|
||||||
|
await asyncio.wait_for(single_image, timeout=10)
|
||||||
|
first_counter = _verify_frame(images[0])
|
||||||
|
|
||||||
|
# Stream request: multiple consecutive frames arrive
|
||||||
|
images.clear()
|
||||||
|
client.request_image_stream()
|
||||||
|
await asyncio.wait_for(stream_done, timeout=10)
|
||||||
|
|
||||||
|
# Frames are distinct, ordered, and fresh per the mock's counter.
|
||||||
|
# Not exactly consecutive: the API drops frames by design while the
|
||||||
|
# previous image is still being sent, so allow small gaps.
|
||||||
|
counters = [_verify_frame(img) for img in images[:STREAM_FRAMES]]
|
||||||
|
for prev, cur in zip(counters, counters[1:], strict=False):
|
||||||
|
assert cur != prev, f"duplicate frames: {counters}"
|
||||||
|
assert ((cur - prev) & 0xFF) < 16, f"frames out of order: {counters}"
|
||||||
|
assert counters[0] != first_counter, "stream should produce new frames"
|
||||||
Reference in New Issue
Block a user