feat: lazy-start picamera2 encoders to save idle CPU

Previously, csi.py started both the MJPEG encoder and (when WebRTC was
enabled) the H264 encoder unconditionally at server startup, regardless
of whether any client was connected. Both encoders ran continuously,
burning CPU on every frame even with zero consumers.

This change introduces LazyEncoder, a small ref-counted wrapper around
picamera2's start_encoder/stop_encoder. Encoders are now started on the
first consumer (MJPEG /stream or /snapshot, WebRTC WHEP POST) and
stopped when the last consumer disconnects.

Measured CPU impact on a Raspberry Pi 4B with a Camera Module 3
(hardware encoders enabled, sampled with `top`, single python process,
percentages are of one core; the kernel reports values >100% when the
process spans multiple cores):

  Default config (640x480 @ 15 FPS):
    Idle (no clients):              ~13-15%  ->  ~5%
    Live MJPEG / WebRTC client:     unchanged

  Heavier config (1920x1080 @ 30 FPS):
    Idle (no clients):              ~95-100%  ->  ~15%
    Live WebRTC only:              ~120-135%  ->  ~60%
    Live MJPEG only:                ~95-104%  ->  ~50-60%

The biggest wins come from two effects: idle is no longer paying for
either encoder, and a live WebRTC-only client no longer pays for the
unused MJPEG encoder running in parallel (and vice versa).

The remaining idle CPU is the picamera2 capture loop itself, which
still runs continuously; pausing the camera entirely is left for a
follow-up commit.

Implementation notes:
- LazyEncoder.acquire/release are protected by a threading.Lock; a
  failed start_encoder rolls back the refcount so callers can retry.
- WebRTC release is wired to RTCPeerConnection 'closed' state with an
  idempotent guard, plus a try/except around setup so a failure during
  SDP exchange or ICE gathering does not leak an encoder.
- USB camera path is unaffected: it never used encoders, and the new
  code uses getattr(handler, ..., None) so missing attributes fall
  back to no-op.

All 129 existing unit tests pass.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: antoinecellerier

spyglass/camera/csi.py

@@ -5,6 +5,7 @@
 from picamera2.outputs import FileOutput
 
 from spyglass import WEBRTC_ENABLED, camera
+from spyglass.camera.lazy_encoder import LazyEncoder
 from spyglass.server.http_server import StreamingHandler
 
 
@@ -41,11 +42,17 @@ def get_frame(inner_self):
                 output.condition.wait()
                 return output.frame
 
-        self.picam2.start_encoder(MJPEGEncoder(), FileOutput(output))
+        StreamingHandler.mjpeg_encoder = LazyEncoder(
+            self.picam2, MJPEGEncoder, FileOutput(output)
+        )
         if WEBRTC_ENABLED:
             from picamera2.encoders import H264Encoder
 
-            self.picam2.start_encoder(H264Encoder(), self.media_track)
+            StreamingHandler.h264_encoder = LazyEncoder(
+                self.picam2, H264Encoder, self.media_track
+            )
+        else:
+            StreamingHandler.h264_encoder = None
         self.picam2.start()
 
         self._run_server(

spyglass/camera/lazy_encoder.py

@@ -0,0 +1,51 @@
+"""Reference-counted lazy start/stop wrapper for picamera2 encoders.
+
+The encoder is only running while at least one consumer holds a reference.
+This avoids burning CPU on encoders that have no clients.
+"""
+
+import threading
+
+
+class LazyEncoder:
+    def __init__(self, picam2, encoder_factory, output):
+        """
+        :param picam2: the Picamera2 instance to start/stop the encoder on.
+        :param encoder_factory: zero-arg callable returning a fresh Encoder.
+        :param output: the picamera2 Output to attach to the encoder.
+        """
+        self._picam2 = picam2
+        self._encoder_factory = encoder_factory
+        self._output = output
+        self._encoder = None
+        self._refs = 0
+        self._lock = threading.Lock()
+
+    def acquire(self):
+        with self._lock:
+            self._refs += 1
+            if self._refs == 1:
+                try:
+                    self._encoder = self._encoder_factory()
+                    self._picam2.start_encoder(self._encoder, self._output)
+                except Exception:
+                    # Roll back so a future caller can retry.
+                    self._refs -= 1
+                    self._encoder = None
+                    raise
+
+    def release(self):
+        with self._lock:
+            if self._refs == 0:
+                return
+            self._refs -= 1
+            if self._refs == 0 and self._encoder is not None:
+                self._picam2.stop_encoder(self._encoder)
+                self._encoder = None
+
+    def __enter__(self):
+        self.acquire()
+        return self
+
+    def __exit__(self, *exc):
+        self.release()

spyglass/server/jpeg.py

@@ -10,6 +10,9 @@
 
 
 def start_streaming(handler: "StreamingHandler"):
+    encoder = getattr(handler, "mjpeg_encoder", None)
+    if encoder is not None:
+        encoder.acquire()
     try:
         send_default_headers(handler)
         handler.send_header("Content-Type", "multipart/x-mixed-replace; boundary=FRAME")
@@ -30,9 +33,15 @@ def start_streaming(handler: "StreamingHandler"):
         logger.warning(
             "Removed streaming client %s: %s", handler.client_address, str(e)
         )
+    finally:
+        if encoder is not None:
+            encoder.release()
 
 
 def send_snapshot(handler: "StreamingHandler"):
+    encoder = getattr(handler, "mjpeg_encoder", None)
+    if encoder is not None:
+        encoder.acquire()
     try:
         send_default_headers(handler)
         frame = handler.get_frame()
@@ -45,6 +54,9 @@ def send_snapshot(handler: "StreamingHandler"):
             handler.wfile.write(frame[2:])
     except Exception as e:
         logger.warning("Removed client %s: %s", handler.client_address, str(e))
+    finally:
+        if encoder is not None:
+            encoder.release()
 
 
 def send_default_headers(handler: "StreamingHandler"):

spyglass/server/webrtc_whep.py

@@ -77,30 +77,48 @@ async def do_POST_async(handler: "StreamingHandler"):
     pc = RTCPeerConnection()
     secret = uuid.uuid4()
 
+    h264_encoder = getattr(handler, "h264_encoder", None)
+    if h264_encoder is not None:
+        h264_encoder.acquire()
+    encoder_released = False
+
+    def _release_encoder_once():
+        nonlocal encoder_released
+        if h264_encoder is not None and not encoder_released:
+            encoder_released = True
+            h264_encoder.release()
+
     @pc.on("connectionstatechange")
     async def on_connectionstatechange():
         print(f"Connection state {pc.connectionState}")
         if pc.connectionState == "failed":
             await pc.close()
         elif pc.connectionState == "closed":
-            pcs.pop(str(secret))
+            pcs.pop(str(secret), None)
+            _release_encoder_once()
             print(f"{len(pcs)} connections still open.")
 
     pcs[str(secret)] = pc
-    track = media_relay.subscribe(handler.media_track)
-    sender = pc.addTrack(track)
-    codecs = RTCRtpSender.getCapabilities("video").codecs
-    transceiver = next(t for t in pc.getTransceivers() if t.sender == sender)
-    transceiver.setCodecPreferences(
-        [codec for codec in codecs if codec.mimeType == "video/H264"]
-    )
-
-    await pc.setRemoteDescription(offer)
-    answer = await pc.createAnswer()
-    await pc.setLocalDescription(answer)
+    try:
+        track = media_relay.subscribe(handler.media_track)
+        sender = pc.addTrack(track)
+        codecs = RTCRtpSender.getCapabilities("video").codecs
+        transceiver = next(t for t in pc.getTransceivers() if t.sender == sender)
+        transceiver.setCodecPreferences(
+            [codec for codec in codecs if codec.mimeType == "video/H264"]
+        )
 
-    while pc.iceGatheringState != "complete":
-        await asyncio.sleep(1)
+        await pc.setRemoteDescription(offer)
+        answer = await pc.createAnswer()
+        await pc.setLocalDescription(answer)
+
+        while pc.iceGatheringState != "complete":
+            await asyncio.sleep(1)
+    except Exception:
+        pcs.pop(str(secret), None)
+        _release_encoder_once()
+        await pc.close()
+        raise
 
     send_default_headers(HTTPStatus.CREATED, handler)