Merge pull request #2723 from bitbarista/fix/camera-rotation-finish-photo-timelapse

fix(camera): apply camera_rotation to finish photos and layer-timelapse frames

backend/app/main.py

@@ -347,6 +347,12 @@ _active_prints: dict[tuple[int, str], int] = {}
 # captures the better-framed pre-bed-drop moment without us having to force
 # timelapse on at dispatch (the #1397 mechanism that caused #1721's per-layer
 # nozzle parking on slicer profiles with Timelapse Type = Smooth).
+#
+# #2708: the bytes in here are ALWAYS already rotated by the printer's
+# camera_rotation. `on_finish_photo_moment` owns that, because one of its
+# sources (the #1867 in-print bank) is rotated before it ever reaches the
+# bank and the others are not — so the consumer can't tell them apart and
+# must not rotate again.
 _stage22_finish_frames: dict[int, bytes] = {}
 
 # #1790: per-printer producer-done event. Set by `on_finish_photo_moment` in its
@@ -1052,6 +1058,7 @@ def _maybe_start_layer_timelapse(printer, printer_id: int, archive_id: int) -> b
         printer.external_camera_url,
         printer.external_camera_type or "mjpeg",
         snapshot_url=printer.external_camera_snapshot_url,
+        rotation=getattr(printer, "camera_rotation", 0),
     )
     logging.getLogger(__name__).info("Started layer timelapse for printer %s, archive %s", printer_id, archive_id)
     return True
@@ -2321,26 +2328,9 @@ async def _maybe_bank_inprint_frame(printer_id: int, layer_num: int) -> None:
 
 def _apply_camera_rotation(image_data: bytes, printer, logger) -> bytes:
     """Apply camera rotation to snapshot image if configured."""
-    rotation = getattr(printer, "camera_rotation", 0)
-    if not rotation or rotation == 0:
-        return image_data
+    from backend.app.services.camera import apply_camera_rotation
 
-    try:
-        from io import BytesIO
-
-        from PIL import Image
-
-        img = Image.open(BytesIO(image_data))
-        # PIL rotate is counter-clockwise, so negate for clockwise rotation
-        img = img.rotate(-rotation, expand=True)
-        buf = BytesIO()
-        img.save(buf, format="JPEG", quality=90)
-        rotated = buf.getvalue()
-        logger.info("[SNAPSHOT] Applied %d° rotation: %s → %s bytes", rotation, len(image_data), len(rotated))
-        return rotated
-    except Exception as e:
-        logger.warning("[SNAPSHOT] Failed to apply rotation: %s", e)
-        return image_data
+    return apply_camera_rotation(image_data, getattr(printer, "camera_rotation", 0), logger)
 
 
 async def _send_print_start_notification(
@@ -3998,6 +3988,7 @@ async def _capture_finish_photo_from_timelapse(
     archive_id: int,
     archive_dir: Path,
     timeout: float | None = None,
+    rotation: int = 0,
 ) -> tuple[str | None, bool]:
     """Wait for the per-print timelapse to land on the archive and extract its
     last frame as the finish photo (#1397).
@@ -4017,11 +4008,16 @@ async def _capture_finish_photo_from_timelapse(
     video landed (whether or not extraction worked), because in that case
     waiting longer changes nothing. The caller uses that to decide between
     falling back permanently and scheduling a background upgrade.
+
+    ``rotation`` is the printer's camera_rotation, applied to the extracted
+    still (#2708) so this source agrees with every other finish-photo source.
+    The archived video itself is the printer's own file and is left alone —
+    rotating it would mean re-encoding it.
     """
     import uuid
 
     from backend.app.models.archive import PrintArchive
-    from backend.app.services.camera import extract_video_last_frame
+    from backend.app.services.camera import apply_camera_rotation_to_file, extract_video_last_frame
 
     logger = logging.getLogger(__name__)
 
@@ -4044,6 +4040,7 @@ async def _capture_finish_photo_from_timelapse(
                 filename = f"finish_{timestamp}_{uuid.uuid4().hex[:8]}.jpg"
                 output_path = photos_dir / filename
                 if await extract_video_last_frame(video_path, output_path):
+                    await apply_camera_rotation_to_file(output_path, rotation, logger)
                     logger.info(
                         "[PHOTO-BG] Extracted finish photo from timelapse %s for archive %s",
                         video_path.name,
@@ -4068,7 +4065,7 @@ async def _capture_finish_photo_from_timelapse(
         await asyncio.sleep(poll_interval)
 
 
-async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Path) -> None:
+async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Path, rotation: int = 0) -> None:
     """Add the timelapse's last frame to an archive after the fact (#2704).
 
     The print-complete notification waits only ~60s for the video, because
@@ -4087,7 +4084,7 @@ async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Pat
     logger = logging.getLogger(__name__)
 
     filename, _ = await _capture_finish_photo_from_timelapse(
-        archive_id, archive_dir, timeout=_FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS
+        archive_id, archive_dir, timeout=_FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS, rotation=rotation
     )
     if not filename:
         logger.info("[PHOTO-UPGRADE] No timelapse frame for archive %s; keeping the live grab", archive_id)
@@ -4372,6 +4369,11 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
                 return
 
         frame_bytes: bytes | None = None
+        # #2708: the banked frame arrives already rotated — it comes from
+        # `_capture_snapshot_for_notification`, which rotates before returning.
+        # Every other source below is a raw grab. Tracking which lets us store
+        # exactly one rotation in `_stage22_finish_frames` either way.
+        frame_already_rotated = False
 
         # #1867: on the FINISH-state fallback the End G-code (e.g. SwapMod
         # plate-swap) has already run, so a live grab now captures the swapped
@@ -4383,6 +4385,7 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
             banked = _inprint_frame_bank.get(printer_id)
             if banked:
                 frame_bytes = banked
+                frame_already_rotated = True
                 logger.info(
                     "[FINISH-PHOTO-MOMENT] using banked in-print frame (%d bytes) — "
                     "avoids post-swap live grab on stage-22-less firmware",
@@ -4436,6 +4439,8 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
                     )
 
         if frame_bytes:
+            if not frame_already_rotated:
+                frame_bytes = _apply_camera_rotation(frame_bytes, printer, logger)
             _stage22_finish_frames[printer_id] = frame_bytes
         else:
             logger.warning(
@@ -5323,6 +5328,7 @@ async def on_print_complete(printer_id: int, data: dict):
                 photo_filename, timelapse_still_pending = await _capture_finish_photo_from_timelapse(
                     archive_id=archive_id,
                     archive_dir=archive_dir,
+                    rotation=getattr(printer, "camera_rotation", 0),
                 )
 
             # #1721: replacement framing path — on_finish_photo_moment
@@ -5350,6 +5356,9 @@ async def on_print_complete(printer_id: int, data: dict):
                         )
                 cached_frame = _stage22_finish_frames.pop(printer_id, None)
                 if cached_frame:
+                    # Already rotated by the producer (#2708) — rotating again
+                    # here would undo the fix on the banked-frame path, whose
+                    # bytes reach the cache having been rotated once already.
                     photos_dir = archive_dir / "photos"
                     photos_dir.mkdir(parents=True, exist_ok=True)
                     timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -5384,6 +5393,7 @@ async def on_print_complete(printer_id: int, data: dict):
                             snapshot_url=printer.external_camera_snapshot_url,
                         )
                     if frame_data:
+                        frame_data = _apply_camera_rotation(frame_data, printer, logger)
                         photos_dir = archive_dir / "photos"
                         photos_dir.mkdir(parents=True, exist_ok=True)
                         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -5401,6 +5411,7 @@ async def on_print_complete(printer_id: int, data: dict):
                     if (active_for_printer or active_chamber_for_printer) and buffered_frame:
                         # Use frame from active stream
                         logger.info("[PHOTO-BG] Using buffered frame from active stream")
+                        buffered_frame = _apply_camera_rotation(buffered_frame, printer, logger)
                         photos_dir = archive_dir / "photos"
                         photos_dir.mkdir(parents=True, exist_ok=True)
                         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -5418,6 +5429,7 @@ async def on_print_complete(printer_id: int, data: dict):
                             access_code=printer.access_code,
                             model=printer.model,
                             archive_dir=archive_dir,
+                            rotation=getattr(printer, "camera_rotation", 0),
                         )
 
             # Write phase: attach the photo in a fresh short-lived session.
@@ -5449,7 +5461,9 @@ async def on_print_complete(printer_id: int, data: dict):
             # gallery never lists.
             if timelapse_still_pending:
                 spawn_background_task(
-                    _upgrade_finish_photo_from_timelapse(archive_id, archive_dir),
+                    _upgrade_finish_photo_from_timelapse(
+                        archive_id, archive_dir, rotation=getattr(printer, "camera_rotation", 0)
+                    ),
                     name=f"finish-photo-upgrade-{archive_id}",
                 )
 

backend/app/services/camera.py

@@ -836,12 +836,72 @@ async def extract_video_last_frame(video_path: Path, output_path: Path) -> bool:
         return False
 
 
+def apply_camera_rotation(image_data: bytes, rotation: int, logger: logging.Logger) -> bytes:
+    """Apply a camera_rotation value (degrees clockwise) to a captured JPEG.
+
+    Shared by every capture path that saves a still image (notification
+    snapshots, finish photos, layer-timelapse frames) - previously only
+    wired into the notification-snapshot path, which left finish photos
+    and timelapse videos upside-down whenever camera_rotation was set.
+
+    Returns *image_data* itself (identity, not a copy) when there is nothing
+    to do or the rotate fails; callers that write to disk use that to skip a
+    pointless rewrite.
+    """
+    if not rotation:
+        return image_data
+
+    try:
+        from io import BytesIO
+
+        from PIL import Image
+
+        img = Image.open(BytesIO(image_data))
+        # PIL rotate is counter-clockwise, so negate for clockwise rotation
+        img = img.rotate(-rotation, expand=True)
+        buf = BytesIO()
+        img.save(buf, format="JPEG", quality=90)
+        rotated = buf.getvalue()
+        # Debug, not info: layer-timelapse calls this once per layer, so a tall
+        # print would otherwise put hundreds of lines in the log for something
+        # the surrounding capture already reports at debug level.
+        logger.debug("Applied %d° camera rotation: %s → %s bytes", rotation, len(image_data), len(rotated))
+        return rotated
+    except Exception as e:
+        logger.warning("Failed to apply camera rotation: %s", e)
+        return image_data
+
+
+async def apply_camera_rotation_to_file(path: Path, rotation: int, logger: logging.Logger) -> None:
+    """Rotate a JPEG that has already been written to disk, in place.
+
+    Two finish-photo sources never hold the frame as bytes - ``ffmpeg`` writes
+    the file for them, and they return only a filename - so they can't use
+    ``apply_camera_rotation`` directly. Best-effort: any failure leaves the
+    unrotated file in place, which is what the caller had before.
+    """
+    if not rotation:
+        return
+
+    try:
+        data = await asyncio.to_thread(path.read_bytes)
+        rotated = await asyncio.to_thread(apply_camera_rotation, data, rotation, logger)
+        if rotated is data:
+            # Nothing was done (the rotate failed and returned its input) -
+            # rewriting the same bytes would only risk truncating a good file.
+            return
+        await asyncio.to_thread(path.write_bytes, rotated)
+    except Exception as e:
+        logger.warning("Failed to rotate %s in place: %s", path.name, e)
+
+
 async def capture_finish_photo(
     printer_id: int,
     ip_address: str,
     access_code: str,
     model: str | None,
     archive_dir: Path,
+    rotation: int = 0,
 ) -> str | None:
     """Capture a finish photo and save it to the archive's photos folder.
 
@@ -851,6 +911,9 @@ async def capture_finish_photo(
         access_code: Printer access code
         model: Printer model
         archive_dir: Directory of the archive (where the 3MF is stored)
+        rotation: Printer's configured camera_rotation (degrees clockwise).
+            ffmpeg writes the file directly here, so the rotation is applied
+            to it afterwards rather than to bytes in hand.
 
     Returns:
         Filename of the captured photo, or None if capture failed
@@ -875,6 +938,7 @@ async def capture_finish_photo(
     )
 
     if success:
+        await apply_camera_rotation_to_file(output_path, rotation, logger)
         logger.info("Finish photo saved: %s", filename)
         return filename
     else:

backend/app/services/layer_timelapse.py

@@ -12,6 +12,7 @@ from datetime import datetime
 from pathlib import Path
 
 from backend.app.core.config import settings
+from backend.app.services.camera import apply_camera_rotation
 from backend.app.services.external_camera import capture_frame
 
 logger = logging.getLogger(__name__)
@@ -51,6 +52,7 @@ class TimelapseSession:
     camera_url: str
     camera_type: str
     snapshot_url: str | None = None  # Optional single-frame override; #1177
+    rotation: int = 0  # Printer's configured camera_rotation, degrees clockwise
     last_layer: int = -1
     frame_count: int = 0
     session_id: str = field(default_factory=lambda: datetime.now().strftime("%Y%m%d_%H%M%S"))
@@ -98,6 +100,8 @@ class TimelapseSession:
             else:
                 frame_data = await capture_frame(self.camera_url, self.camera_type, snapshot_url=self.snapshot_url)
             if frame_data:
+                if self.rotation:
+                    frame_data = await asyncio.to_thread(apply_camera_rotation, frame_data, self.rotation, logger)
                 frame_path = self.frames_dir / f"layer_{layer_num:05d}.jpg"
                 await asyncio.to_thread(frame_path.write_bytes, frame_data)
                 self.frame_count += 1
@@ -216,6 +220,7 @@ def start_session(
     url: str,
     cam_type: str,
     snapshot_url: str | None = None,
+    rotation: int = 0,
 ) -> TimelapseSession:
     """Start new timelapse session for a printer.
 
@@ -226,6 +231,8 @@ def start_session(
         cam_type: Camera type ("mjpeg", "rtsp", "snapshot")
         snapshot_url: Optional single-frame URL override; when set, layer captures
             fetch from it directly instead of opening the live stream. #1177.
+        rotation: Printer's configured camera_rotation (degrees clockwise),
+            applied to every captured frame before it's saved.
 
     Returns:
         The new TimelapseSession
@@ -239,6 +246,7 @@ def start_session(
         camera_url=url,
         camera_type=cam_type,
         snapshot_url=snapshot_url,
+        rotation=rotation,
     )
     _active_sessions[printer_id] = session
     logger.info("Started timelapse session for printer %s", printer_id)

backend/tests/unit/services/test_camera_rotation.py

@@ -0,0 +1,146 @@
+"""Tests for the shared camera-rotation helpers (#2708).
+
+Every other test of a rotating path patches ``apply_camera_rotation`` out and
+asserts the call, which proves the wiring but not the rotation. These drive
+the real PIL round trip, so a flipped sign or a dropped ``expand=True`` fails
+here rather than shipping.
+"""
+
+import io
+import logging
+
+import pytest
+from PIL import Image
+
+from backend.app.services.camera import apply_camera_rotation, apply_camera_rotation_to_file
+
+logger = logging.getLogger(__name__)
+
+
+def _jpeg(width: int, height: int, corner: tuple[int, int, int] = (255, 0, 0)) -> bytes:
+    """A JPEG with one distinctly coloured pixel block in the top-left corner,
+    so which way it turned is observable and not just the dimensions."""
+    img = Image.new("RGB", (width, height), (0, 0, 255))
+    for x in range(min(8, width)):
+        for y in range(min(8, height)):
+            img.putpixel((x, y), corner)
+    buf = io.BytesIO()
+    img.save(buf, format="JPEG", quality=95)
+    return buf.getvalue()
+
+
+def _open(data: bytes) -> Image.Image:
+    return Image.open(io.BytesIO(data))
+
+
+def _brightest_corner(img: Image.Image) -> str:
+    """Which corner holds the red block, sampled a few pixels in to stay clear
+    of JPEG ringing at the edges."""
+    w, h = img.size
+    probes = {
+        "top-left": (3, 3),
+        "top-right": (w - 4, 3),
+        "bottom-left": (3, h - 4),
+        "bottom-right": (w - 4, h - 4),
+    }
+    return max(probes, key=lambda name: img.getpixel(probes[name])[0] - img.getpixel(probes[name])[2])
+
+
+class TestApplyCameraRotation:
+    def test_zero_rotation_returns_the_input_object(self):
+        """Not merely equal — identity. apply_camera_rotation_to_file uses this
+        to decide there is nothing to write back."""
+        src = _jpeg(64, 32)
+        assert apply_camera_rotation(src, 0, logger) is src
+
+    def test_90_degrees_turns_clockwise(self):
+        """camera_rotation is documented as degrees *clockwise*, and PIL's
+        rotate() is counter-clockwise — the helper negates to compensate. A
+        lost negation would send the corner to bottom-right instead."""
+        src = _jpeg(64, 32)
+        assert _brightest_corner(_open(src)) == "top-left"
+
+        out = _open(apply_camera_rotation(src, 90, logger))
+        assert out.size == (32, 64)  # expand=True, so the frame is not cropped
+        assert _brightest_corner(out) == "top-right"
+
+    def test_270_degrees_turns_the_other_way(self):
+        out = _open(apply_camera_rotation(_jpeg(64, 32), 270, logger))
+        assert out.size == (32, 64)
+        assert _brightest_corner(out) == "bottom-left"
+
+    def test_180_degrees_keeps_the_dimensions_and_flips_the_corner(self):
+        out = _open(apply_camera_rotation(_jpeg(64, 32), 180, logger))
+        assert out.size == (64, 32)
+        assert _brightest_corner(out) == "bottom-right"
+
+    def test_applying_180_twice_is_the_bug_that_was_fixed(self):
+        """The regression this guards: two rotations cancel out and the photo
+        is upside-down again. Kept as a test so the invariant that
+        _stage22_finish_frames holds exactly one rotation has a stated reason.
+        """
+        src = _jpeg(64, 32)
+        once = apply_camera_rotation(src, 180, logger)
+        twice = apply_camera_rotation(once, 180, logger)
+        assert _brightest_corner(_open(once)) == "bottom-right"
+        assert _brightest_corner(_open(twice)) == "top-left"  # back to the original
+
+    def test_undecodable_bytes_return_unchanged(self):
+        """A capture path must not lose a frame because the rotate failed —
+        an unrotated photo beats no photo."""
+        junk = b"not a jpeg at all"
+        assert apply_camera_rotation(junk, 90, logger) is junk
+
+    def test_a_failed_rotate_is_logged_as_a_warning(self, caplog):
+        with caplog.at_level(logging.WARNING, logger=__name__):
+            apply_camera_rotation(b"not a jpeg at all", 90, logger)
+        assert any("Failed to apply camera rotation" in r.message for r in caplog.records)
+
+    def test_a_successful_rotate_does_not_log_at_info(self, caplog):
+        """Layer-timelapse calls this once per layer; at INFO a tall print
+        would bury the log."""
+        with caplog.at_level(logging.INFO, logger=__name__):
+            apply_camera_rotation(_jpeg(64, 32), 90, logger)
+        assert caplog.records == []
+
+
+class TestApplyCameraRotationToFile:
+    """The two finish-photo sources that let ffmpeg write the file and never
+    hold the bytes: capture_finish_photo and the timelapse last-frame extract."""
+
+    @pytest.mark.asyncio
+    async def test_rotates_in_place(self, tmp_path):
+        path = tmp_path / "finish.jpg"
+        path.write_bytes(_jpeg(64, 32))
+
+        await apply_camera_rotation_to_file(path, 90, logger)
+
+        out = _open(path.read_bytes())
+        assert out.size == (32, 64)
+        assert _brightest_corner(out) == "top-right"
+
+    @pytest.mark.asyncio
+    async def test_zero_rotation_leaves_the_file_untouched(self, tmp_path):
+        path = tmp_path / "finish.jpg"
+        original = _jpeg(64, 32)
+        path.write_bytes(original)
+
+        await apply_camera_rotation_to_file(path, 0, logger)
+
+        assert path.read_bytes() == original
+
+    @pytest.mark.asyncio
+    async def test_a_file_that_cannot_be_rotated_is_left_intact(self, tmp_path):
+        """Not truncated, not deleted — the caller's unrotated photo survives."""
+        path = tmp_path / "finish.jpg"
+        path.write_bytes(b"not a jpeg at all")
+
+        await apply_camera_rotation_to_file(path, 90, logger)
+
+        assert path.read_bytes() == b"not a jpeg at all"
+
+    @pytest.mark.asyncio
+    async def test_a_missing_file_does_not_raise(self, tmp_path):
+        """Best-effort: this runs after the capture reported success, and must
+        not turn a delivered photo into a failed one."""
+        await apply_camera_rotation_to_file(tmp_path / "gone.jpg", 90, logger)

backend/tests/unit/services/test_layer_timelapse.py

@@ -7,7 +7,7 @@ These tests cover session management and pure logic functions.
 import time
 from datetime import datetime
 from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -247,6 +247,99 @@ class TestLayerChangeLogic:
                     assert session.frame_count == 0  # But frame count not incremented
 
 
+class TestCaptureLayerAppliesRotation:
+    """camera_rotation was previously only wired into the notification-
+    snapshot path, so a layer-timelapse video came out upside-down whenever
+    the printer had a rotation configured. capture_layer now applies it to
+    every captured frame, whether fresh or reused from the live view's
+    buffer, before writing to disk."""
+
+    @pytest.mark.asyncio
+    async def test_rotates_fresh_capture_when_configured(self, tmp_path):
+        from backend.app.services.layer_timelapse import TimelapseSession
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+
+            with patch.object(Path, "mkdir"):
+                session = TimelapseSession(1, 100, "/dev/video1", "usb", rotation=180)
+
+                with (
+                    patch("backend.app.api.routes.camera.live_frame_for_capture", return_value=(False, None)),
+                    patch(
+                        "backend.app.services.layer_timelapse.capture_frame",
+                        new_callable=AsyncMock,
+                        return_value=b"\xff\xd8unrotated\xff\xd9",
+                    ),
+                    patch(
+                        "backend.app.services.layer_timelapse.apply_camera_rotation",
+                        return_value=b"\xff\xd8rotated\xff\xd9",
+                    ) as mock_rotate,
+                    patch.object(Path, "write_bytes") as mock_write,
+                ):
+                    result = await session.capture_layer(1)
+
+        assert result is True
+        mock_rotate.assert_called_once_with(b"\xff\xd8unrotated\xff\xd9", 180, ANY)
+        mock_write.assert_called_once_with(b"\xff\xd8rotated\xff\xd9")
+
+    @pytest.mark.asyncio
+    async def test_rotates_buffered_frame_when_configured(self, tmp_path):
+        from backend.app.services.layer_timelapse import TimelapseSession
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+
+            with patch.object(Path, "mkdir"):
+                session = TimelapseSession(1, 100, "/dev/video1", "usb", rotation=90)
+
+                with (
+                    patch(
+                        "backend.app.api.routes.camera.live_frame_for_capture",
+                        return_value=(True, b"\xff\xd8buffered\xff\xd9"),
+                    ),
+                    patch(
+                        "backend.app.services.layer_timelapse.apply_camera_rotation",
+                        return_value=b"\xff\xd8rotated\xff\xd9",
+                    ) as mock_rotate,
+                    patch.object(Path, "write_bytes") as mock_write,
+                ):
+                    result = await session.capture_layer(1)
+
+        assert result is True
+        mock_rotate.assert_called_once_with(b"\xff\xd8buffered\xff\xd9", 90, ANY)
+        mock_write.assert_called_once_with(b"\xff\xd8rotated\xff\xd9")
+
+    @pytest.mark.asyncio
+    async def test_skips_rotation_when_not_configured(self, tmp_path):
+        """Default rotation=0 - no-op, and must not even call apply_camera_rotation
+        (avoids the PIL decode/re-encode round trip for the common case)."""
+        from backend.app.services.layer_timelapse import TimelapseSession
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+
+            with patch.object(Path, "mkdir"):
+                session = TimelapseSession(1, 100, "/dev/video1", "usb")
+                assert session.rotation == 0
+
+                with (
+                    patch("backend.app.api.routes.camera.live_frame_for_capture", return_value=(False, None)),
+                    patch(
+                        "backend.app.services.layer_timelapse.capture_frame",
+                        new_callable=AsyncMock,
+                        return_value=b"\xff\xd8unrotated\xff\xd9",
+                    ),
+                    patch("backend.app.services.layer_timelapse.apply_camera_rotation") as mock_rotate,
+                    patch.object(Path, "write_bytes") as mock_write,
+                ):
+                    result = await session.capture_layer(1)
+
+        assert result is True
+        mock_rotate.assert_not_called()
+        mock_write.assert_called_once_with(b"\xff\xd8unrotated\xff\xd9")
+
+
 class TestOnLayerChange:
     """Tests for the on_layer_change callback."""
 

backend/tests/unit/test_finish_photo_from_timelapse.py

@@ -179,3 +179,69 @@ async def test_polls_until_file_appears(tmp_path: Path, patched_session, monkeyp
 
     assert result is not None
     assert result.startswith("finish_")
+
+
+async def test_extracted_frame_is_rotated_when_configured(tmp_path: Path, patched_session, monkeypatch):
+    """#2708: this source hands a path to ffmpeg and never holds the bytes, so
+    it was the one finish-photo source that ignored camera_rotation entirely.
+    A built-in-camera print with a timelapse prefers this source over the live
+    grab, so leaving it out meant the orientation depended on which source won.
+    """
+    import io
+
+    from PIL import Image
+
+    monkeypatch.setattr(main_module.app_settings, "base_dir", tmp_path)
+    video_relpath = Path("archive/1/print/timelapse.mp4")
+    video_abspath = tmp_path / video_relpath
+    video_abspath.parent.mkdir(parents=True, exist_ok=True)
+    video_abspath.write_bytes(b"x" * 100)
+
+    async def fake_extract(src, dst):
+        buf = io.BytesIO()
+        Image.new("RGB", (64, 32), (0, 0, 255)).save(buf, format="JPEG")
+        dst.write_bytes(buf.getvalue())
+        return True
+
+    monkeypatch.setattr(main_module, "_FINISH_PHOTO_TIMELAPSE_POLL_INTERVAL_SECONDS", 0.0)
+    patched_session.timelapse_path = str(video_relpath)
+
+    with patch("backend.app.services.camera.extract_video_last_frame", new=fake_extract):
+        result, _ = await _capture_finish_photo_from_timelapse(
+            archive_id=42,
+            archive_dir=tmp_path / "archive_dir",
+            rotation=90,
+        )
+
+    assert result is not None
+    written = tmp_path / "archive_dir" / "photos" / result
+    # 64x32 turned a quarter turn: the file on disk is the rotated one, not
+    # what ffmpeg wrote.
+    assert Image.open(io.BytesIO(written.read_bytes())).size == (32, 64)
+
+
+async def test_extracted_frame_is_untouched_without_a_rotation(tmp_path: Path, patched_session, monkeypatch):
+    """The default path must not decode and re-encode ffmpeg's output for
+    nothing — that would cost a generation of JPEG quality on every print."""
+    monkeypatch.setattr(main_module.app_settings, "base_dir", tmp_path)
+    video_relpath = Path("archive/1/print/timelapse.mp4")
+    video_abspath = tmp_path / video_relpath
+    video_abspath.parent.mkdir(parents=True, exist_ok=True)
+    video_abspath.write_bytes(b"x" * 100)
+
+    extracted = b"\xff\xd8" + b"\x00" * 50
+
+    async def fake_extract(src, dst):
+        dst.write_bytes(extracted)
+        return True
+
+    monkeypatch.setattr(main_module, "_FINISH_PHOTO_TIMELAPSE_POLL_INTERVAL_SECONDS", 0.0)
+    patched_session.timelapse_path = str(video_relpath)
+
+    with patch("backend.app.services.camera.extract_video_last_frame", new=fake_extract):
+        result, _ = await _capture_finish_photo_from_timelapse(
+            archive_id=42,
+            archive_dir=tmp_path / "archive_dir",
+        )
+
+    assert (tmp_path / "archive_dir" / "photos" / result).read_bytes() == extracted

backend/tests/unit/test_finish_photo_moment_sync.py

@@ -322,3 +322,139 @@ async def test_bank_skips_during_calibration_substage(monkeypatch):
     _bank_env(monkeypatch, sub_stage=14)
     await main_module._maybe_bank_inprint_frame(3, 2)
     assert 3 not in main_module._inprint_frame_bank
+
+
+class TestStage22CacheHoldsExactlyOneRotation:
+    """#2708. `_stage22_finish_frames` is fed from two kinds of source: live
+    grabs, which are raw, and the #1867 in-print bank, whose bytes came from
+    `_capture_snapshot_for_notification` and are therefore ALREADY rotated.
+    The consumer cannot tell them apart, so the producer normalises: every
+    entry in the cache has had the rotation applied exactly once.
+
+    Rotating on the consumer side instead put two rotations on the banked
+    path — at 180 degrees that is the reported bug reproduced exactly, and at
+    90/270 it lands the photo 180 degrees out.
+    """
+
+    @staticmethod
+    def _jpeg(width, height):
+        import io
+
+        from PIL import Image
+
+        buf = io.BytesIO()
+        Image.new("RGB", (width, height), (0, 0, 255)).save(buf, format="JPEG")
+        return buf.getvalue()
+
+    @staticmethod
+    def _size(data):
+        import io
+
+        from PIL import Image
+
+        return Image.open(io.BytesIO(data)).size
+
+    async def test_a_live_grab_is_rotated_before_caching(self, patched_env, monkeypatch):
+        monkeypatch.setattr(patched_env, "camera_rotation", 90, raising=False)
+        raw = self._jpeg(64, 32)
+
+        async def _capture(**_kwargs):
+            return raw
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
+
+        cached = main_module._stage22_finish_frames[patched_env.id]
+        assert self._size(cached) == (32, 64)
+
+    async def test_the_banked_frame_is_cached_verbatim(self, patched_env, monkeypatch):
+        """The bank is filled by `_capture_snapshot_for_notification`, which
+        rotates before it returns — so the producer must pass those bytes
+        through untouched rather than rotating them a second time.
+
+        Note this pins the invariant forward; it does not on its own prove the
+        bug fixed, because the old producer didn't rotate anything either. The
+        pair that discriminates is `test_a_live_grab_is_rotated_before_caching`
+        (producer now rotates) plus the source guard below (consumer no longer
+        does).
+        """
+        monkeypatch.setattr(patched_env, "camera_rotation", 90, raising=False)
+        already_rotated = self._jpeg(32, 64)  # what one rotation of a 64x32 frame looks like
+        main_module._inprint_frame_bank[patched_env.id] = already_rotated
+
+        async def _capture(**_kwargs):  # pragma: no cover - must not be reached
+            raise AssertionError("the banked frame should have been preferred")
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
+
+        cached = main_module._stage22_finish_frames[patched_env.id]
+        assert cached is already_rotated
+        assert self._size(cached) == (32, 64)
+
+    async def test_a_stage22_grab_is_rotated_even_though_the_bank_is_full(self, patched_env, monkeypatch):
+        """Only the `finish_state` trigger reads the bank. The `stage_22` and
+        `last_layer` triggers take a live grab, which still needs rotating —
+        a shared "did we use the bank" flag must not latch on the bank merely
+        existing."""
+        monkeypatch.setattr(patched_env, "camera_rotation", 90, raising=False)
+        main_module._inprint_frame_bank[patched_env.id] = self._jpeg(999, 1)
+
+        async def _capture(**_kwargs):
+            return self._jpeg(64, 32)
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "stage_22"})
+
+        cached = main_module._stage22_finish_frames[patched_env.id]
+        assert self._size(cached) == (32, 64)
+
+    async def test_no_rotation_configured_caches_the_bytes_as_captured(self, patched_env, monkeypatch):
+        raw = self._jpeg(64, 32)
+
+        async def _capture(**_kwargs):
+            return raw
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
+
+        assert main_module._stage22_finish_frames[patched_env.id] is raw
+
+
+def test_the_consumer_does_not_rotate_the_cached_frame():
+    """The other half of the #2708 invariant, and the half with no runtime
+    harness: `_background_finish_photo` is a closure nested inside
+    `on_print_complete`, so nothing can drive its cached-frame branch
+    directly. What it must NOT do is rotate what it pops from
+    `_stage22_finish_frames` — the producer has already done that, and doing
+    it again upside-downs the banked path, which is the bug this fixed.
+
+    Checked against the source because the alternative is no check at all.
+    """
+    import ast
+    from pathlib import Path
+
+    main_py = Path(__file__).resolve().parents[2] / "app" / "main.py"
+    assert main_py.exists(), f"guard is looking in the wrong place: {main_py}"
+    tree = ast.parse(main_py.read_text())
+
+    offenders = [
+        node.lineno
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_apply_camera_rotation"
+        and node.args
+        and isinstance(node.args[0], ast.Name)
+        and node.args[0].id == "cached_frame"
+    ]
+
+    assert not offenders, (
+        f"main.py:{offenders} rotates the frame popped from _stage22_finish_frames. "
+        "Those bytes are already rotated by on_finish_photo_moment (#2708); rotating "
+        "again returns a 180-degree print to upside-down."
+    )

backend/tests/unit/test_layer_timelapse_expected_archive.py

@@ -60,6 +60,7 @@ def test_starts_timelapse_when_external_camera_enabled():
         "http://camera.local:5000/snapshot.jpg",
         "snapshot",
         snapshot_url="http://camera.local:5000/snapshot.jpg",
+        rotation=0,
     )