Update hardware tests and fix PNG saving with picamera

This commit is contained in:
Julian Stirling 2026-06-03 16:56:41 +01:00
parent e3e3afbca6
commit 29518e9cb9
5 changed files with 68 additions and 20 deletions

View file

@ -42,6 +42,10 @@ class ImageFormatInfo(BaseModel):
supported_extensions: tuple[str, ...] supported_extensions: tuple[str, ...]
"""All supported extension (lowercase).""" """All supported extension (lowercase)."""
def path_matches(self, path: str) -> bool:
"""Return True if path matches one of the supported extensions."""
return path.lower().endswith(self.supported_extensions)
BASE_IMAGE_FORMATS: dict[str, ImageFormatInfo] = { BASE_IMAGE_FORMATS: dict[str, ImageFormatInfo] = {
"jpeg": ImageFormatInfo( "jpeg": ImageFormatInfo(
@ -462,7 +466,7 @@ class BaseCamera(OFMThing, ABC):
complete, this error will not be raised until the data is accessed. Consider complete, this error will not be raised until the data is accessed. Consider
using ``grab_jpeg_as_array`` instead. using ``grab_jpeg_as_array`` instead.
This differs from ``capture_jpeg`` in that it does not pause the MJPEG This differs from ``capture`` in that it does not pause the MJPEG
preview stream. Instead, we simply return the next frame from that preview stream. Instead, we simply return the next frame from that
stream (either "main" for the preview stream, or "lores" for the low stream (either "main" for the preview stream, or "lores" for the low
resolution preview). No metadata is returned. resolution preview). No metadata is returned.
@ -679,8 +683,7 @@ class BaseCamera(OFMThing, ABC):
image = image.resize(save_resolution, Image.Resampling.BOX) image = image.resize(save_resolution, Image.Resampling.BOX)
try: try:
save_kwargs: dict[str, Any] = {} save_kwargs: dict[str, Any] = {}
jpeg_exts = BASE_IMAGE_FORMATS["jpeg"].supported_extensions if BASE_IMAGE_FORMATS["jpeg"].path_matches(resolved_path):
if resolved_path.lower().endswith(jpeg_exts):
# Per PIL documentation, # Per PIL documentation,
# (https://pillow.readthedocs.io/en/stable/handbook/image-file-formats.html#jpeg) # (https://pillow.readthedocs.io/en/stable/handbook/image-file-formats.html#jpeg)
# there are two factors when saving a JPEG. Subsampling affects the colour, # there are two factors when saving a JPEG. Subsampling affects the colour,
@ -689,6 +692,11 @@ class BaseCamera(OFMThing, ABC):
# quality = 95 is the maximum recommended - above this, JPEG compression is # quality = 95 is the maximum recommended - above this, JPEG compression is
# disabled, file size increases and quality is barely or not affected # disabled, file size increases and quality is barely or not affected
save_kwargs = {"quality": 95, "subsampling": 0} save_kwargs = {"quality": 95, "subsampling": 0}
if (
BASE_IMAGE_FORMATS["png"].path_matches(resolved_path)
and image.mode == "RGBX"
):
image = image.convert("RGB")
image.save(resolved_path, **save_kwargs) image.save(resolved_path, **save_kwargs)
try: try:
self._add_metadata_to_capture(resolved_path, dict(metadata)) self._add_metadata_to_capture(resolved_path, dict(metadata))

View file

@ -7,11 +7,8 @@ import numpy as np
from PIL import Image from PIL import Image
def test_jpeg_and_array(picamera_client): def test_quick_capture_size(picamera_client):
"""Check that a jpeg grabbed from the stream is the same size as other captures. """Check that a jpeg grabbed from the stream is the same size as a quick capture."""
Compare it to an array capture and a jpeg capture.
"""
# Grab a jpeg from the stream # Grab a jpeg from the stream
blob = picamera_client.grab_jpeg() blob = picamera_client.grab_jpeg()
mjpeg_frame = Image.open(blob.open()) mjpeg_frame = Image.open(blob.open())
@ -20,13 +17,17 @@ def test_jpeg_and_array(picamera_client):
assert mjpeg_frame.format == "JPEG" assert mjpeg_frame.format == "JPEG"
# Capture a jpeg # Capture a jpeg
blob = picamera_client.capture_jpeg(stream_name="main") blob = picamera_client.capture(
capture_mode="quick",
image_format="jpeg",
retain_image=True,
)
jpeg_capture = Image.open(blob.open()) jpeg_capture = Image.open(blob.open())
jpeg_capture.verify() jpeg_capture.verify()
assert jpeg_capture.format == "JPEG" assert jpeg_capture.format == "JPEG"
# Capture an array # Capture an array
arrlist = picamera_client.capture_as_array(stream_name="main") arrlist = picamera_client.capture_as_array(capture_mode="quick")
array_main = np.array(arrlist) array_main = np.array(arrlist)
# Verify image sizes are the same # Verify image sizes are the same
@ -34,6 +35,41 @@ def test_jpeg_and_array(picamera_client):
assert array_main.shape[1::-1] == jpeg_capture.size assert array_main.shape[1::-1] == jpeg_capture.size
def test_format(picamera_client):
"""Check capture format is as requested."""
# Capture a jpeg
blob = picamera_client.capture(
capture_mode="quick",
image_format="jpeg",
retain_image=True,
)
jpeg_capture = Image.open(blob.open())
jpeg_capture.verify()
assert jpeg_capture.format == "JPEG"
blob = picamera_client.capture(
capture_mode="quick",
image_format="png",
retain_image=True,
)
jpeg_capture = Image.open(blob.open())
jpeg_capture.verify()
assert jpeg_capture.format == "PNG"
def test_standard_capture_size(picamera_client):
"""Check standard capture mode captures at expected size."""
# Capture a jpeg
blob = picamera_client.capture(
capture_mode="standard",
image_format="jpeg",
retain_image=True,
)
jpeg_capture = Image.open(blob.open())
jpeg_capture.verify()
assert jpeg_capture.size == (1640, 1232)
def test_record_framerate(picamera_client): def test_record_framerate(picamera_client):
"""Check that framerate monitoring creates a valid JSON log with good data.""" """Check that framerate monitoring creates a valid JSON log with good data."""
log_file = Path(picamera_client.record_framerate(duration=1.0)) log_file = Path(picamera_client.record_framerate(duration=1.0))

View file

@ -39,7 +39,7 @@ def _test_exposure_time_drift(desired_time: int) -> None:
assert abs(pre_capture_et - desired_time) < EXPOSURE_TOL assert abs(pre_capture_et - desired_time) < EXPOSURE_TOL
for i in range(10): for i in range(10):
client.capture_jpeg(stream_name="full") client.capture(capture_mode="full")
if i == 0: if i == 0:
# Exposure can update on first capture, due to frame rate restrictions # Exposure can update on first capture, due to frame rate restrictions
first_et = client.exposure_time first_et = client.exposure_time
@ -56,7 +56,7 @@ def _test_exposure_time_drift(desired_time: int) -> None:
time.sleep(0.5) time.sleep(0.5)
# Check before and after capture # Check before and after capture
assert client.exposure_time == frame_et assert client.exposure_time == frame_et
client.capture_jpeg(stream_name="full") client.capture(capture_mode="full")
assert client.exposure_time == frame_et assert client.exposure_time == frame_et
print("Exposure time didn't change!!") print("Exposure time didn't change!!")
print(f"End of test for exposure target {desired_time}") print(f"End of test for exposure target {desired_time}")
@ -82,7 +82,7 @@ def test_exposure_time_on_start_and_stop_stream():
# Take a couple of images to make sure that the exposure is adjusted to # Take a couple of images to make sure that the exposure is adjusted to
# a hardware compatible value. # a hardware compatible value.
for _i in range(2): for _i in range(2):
client.capture_jpeg(stream_name="full") client.capture(capture_mode="full")
# Save this time. # Save this time.
set_time = client.exposure_time set_time = client.exposure_time
assert abs(set_time - desired_time) < EXPOSURE_TOL assert abs(set_time - desired_time) < EXPOSURE_TOL
@ -112,7 +112,7 @@ def _load_camera_and_return_exposure(tmpdir: str) -> int:
# Take a couple of images to make sure that the exposure is adjusted to # Take a couple of images to make sure that the exposure is adjusted to
# a hardware compatible value. # a hardware compatible value.
for _i in range(2): for _i in range(2):
client.capture_jpeg(stream_name="full") client.capture(capture_mode="full")
# Save this time. # Save this time.
return client.exposure_time return client.exposure_time

View file

@ -14,7 +14,7 @@ def test_streaming_mode():
with camera_test_client() as client: with camera_test_client() as client:
for mode, res in ["default", (820, 616)], ["full_resolution", (3280, 2464)]: for mode, res in ["default", (820, 616)], ["full_resolution", (3280, 2464)]:
client.change_streaming_mode(mode=mode) client.change_streaming_mode(mode=mode)
arr = np.array(client.capture_as_array(stream_name="main")) arr = np.array(client.capture_as_array(capture_mode="quick"))
# Check that the array dimensions match the requested image size. # Check that the array dimensions match the requested image size.
# Note: Numpy array shape is (y,x), but the sensor is set with (x,y) # Note: Numpy array shape is (y,x), but the sensor is set with (x,y)
# hence the need to compare index 0 with index 1. # hence the need to compare index 0 with index 1.

View file

@ -79,6 +79,7 @@ class MemorySaveTestCase:
filename: str = "foobar.jpeg" filename: str = "foobar.jpeg"
save_resolution: Optional[tuple[int, int]] = None save_resolution: Optional[tuple[int, int]] = None
resize_needed: bool = False resize_needed: bool = False
convert_needed: bool = False
save_kwargs: dict[str, int] = field( save_kwargs: dict[str, int] = field(
default_factory=lambda: {"quality": 95, "subsampling": 0} default_factory=lambda: {"quality": 95, "subsampling": 0}
) )
@ -91,9 +92,9 @@ SAVE_TEST_CASES = [
MemorySaveTestCase("foobar.JPEG"), MemorySaveTestCase("foobar.JPEG"),
MemorySaveTestCase("foobar.JPG"), MemorySaveTestCase("foobar.JPG"),
MemorySaveTestCase("foobar.png.jpeg"), MemorySaveTestCase("foobar.png.jpeg"),
MemorySaveTestCase("foobar.png", save_kwargs={}), MemorySaveTestCase("foobar.png", save_kwargs={}, convert_needed=True),
MemorySaveTestCase("foobar.PNG", save_kwargs={}), MemorySaveTestCase("foobar.PNG", save_kwargs={}, convert_needed=True),
MemorySaveTestCase("foobar.jpeg.png", save_kwargs={}), MemorySaveTestCase("foobar.jpeg.png", save_kwargs={}, convert_needed=True),
MemorySaveTestCase(save_resolution=None, resize_needed=False), MemorySaveTestCase(save_resolution=None, resize_needed=False),
MemorySaveTestCase(save_resolution=(1000, 1200), resize_needed=False), MemorySaveTestCase(save_resolution=(1000, 1200), resize_needed=False),
MemorySaveTestCase(save_resolution=(2000, 2400), resize_needed=True), MemorySaveTestCase(save_resolution=(2000, 2400), resize_needed=True),
@ -112,10 +113,12 @@ def test_save_from_memory(test_case, test_env, mocker):
mocker.patch.object(type(camera), "capture_modes", capture_modes_mock) mocker.patch.object(type(camera), "capture_modes", capture_modes_mock)
mock_image = mocker.Mock() mock_image = mocker.Mock()
# Make resize return itself so we can track further calls of the Image object after # Make resize and convert return itself so we can track further calls of the Image
# a resize # object after a resize
mock_image.resize.return_value = mock_image mock_image.resize.return_value = mock_image
mock_image.convert.return_value = mock_image
mock_image.size = (1000, 1200) mock_image.size = (1000, 1200)
mock_image.mode = "RGBX"
camera._memory_buffer.get_image.return_value = ( camera._memory_buffer.get_image.return_value = (
mock_image, mock_image,
@ -131,5 +134,6 @@ def test_save_from_memory(test_case, test_env, mocker):
assert camera._memory_buffer.get_image.call_args.args == (33,) assert camera._memory_buffer.get_image.call_args.args == (33,)
assert camera._add_metadata_to_capture.call_count == 1 assert camera._add_metadata_to_capture.call_count == 1
assert mock_image.resize.call_count == (1 if test_case.resize_needed else 0) assert mock_image.resize.call_count == (1 if test_case.resize_needed else 0)
assert mock_image.convert.call_count == (1 if test_case.convert_needed else 0)
assert mock_image.save.call_count == 1 assert mock_image.save.call_count == 1
assert mock_image.save.call_args.kwargs == test_case.save_kwargs assert mock_image.save.call_args.kwargs == test_case.save_kwargs