from __future__ import annotations import io import logging import shutil import subprocess import threading import time from datetime import UTC, datetime from pathlib import Path logger = logging.getLogger(__name__) try: from picamera2 import Picamera2 from picamera2.encoders import H264Encoder from picamera2.outputs import FileOutput PICAMERA_AVAILABLE = True except ImportError: PICAMERA_AVAILABLE = False HLS_DIR = Path("/tmp/hls") RECORDINGS_DIR = Path("/var/lib/birdcam/recordings") SEGMENT_DURATION = 2 # seconds per segment SEGMENT_COUNT = 5 # segments to keep in playlist BITRATE = 2_000_000 # 2 Mbps — adjust for bandwidth needs class PipeOutput(io.RawIOBase): """Wraps ffmpeg stdin pipe as a BufferedIOBase-compatible stream.""" def __init__(self, proc: subprocess.Popen[bytes]) -> None: self._proc = proc def write(self, data: bytes) -> int: # type: ignore[override] if self._proc.stdin and not self._proc.stdin.closed: try: self._proc.stdin.write(data) return len(data) except BrokenPipeError: pass return 0 def writable(self) -> bool: return True class Camera: def __init__(self) -> None: self._picam: Picamera2 | None = None self._stream_encoder: H264Encoder | None = None self._record_encoder: H264Encoder | None = None self._ffmpeg: subprocess.Popen[bytes] | None = None self._record_ffmpeg: subprocess.Popen[bytes] | None = None self._output: PipeOutput | None = None self._ready_event = threading.Event() self._watch_thread: threading.Thread | None = None self._stop_event = threading.Event() self.running = False self.recording = False self._current_recording_path: Path | None = None self._recording_started_at: datetime | None = None self._lock = threading.Lock() # ── Streaming ──────────────────────────────────────────────────────────── def start(self) -> None: if self.running: return # prepare HLS output directory if HLS_DIR.exists(): shutil.rmtree(HLS_DIR) HLS_DIR.mkdir(parents=True) if not PICAMERA_AVAILABLE: logger.info("Mock camera started (picamera2 not available)") self.running = True return # start ffmpeg: reads raw H.264 from stdin, writes HLS segments ffmpeg_cmd = [ "ffmpeg", "-loglevel", "warning", "-f", "h264", "-i", "pipe:0", "-c:v", "copy", "-hls_time", str(SEGMENT_DURATION), "-hls_list_size", str(SEGMENT_COUNT), "-hls_flags", "delete_segments+append_list", "-hls_segment_filename", str(HLS_DIR / "seg%03d.ts"), str(HLS_DIR / "stream.m3u8"), ] self._ffmpeg = subprocess.Popen( ffmpeg_cmd, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) self._output = PipeOutput(self._ffmpeg) self._picam = Picamera2() config = self._picam.create_video_configuration( main={"size": (1280, 720)}, ) self._picam.configure(config) self._picam.set_controls( { "Brightness": 0.1, "Contrast": 1.1, "Saturation": 1.1, "Sharpness": 1.0, "AwbEnable": True, "AeEnable": True, } ) self._stream_encoder = H264Encoder(bitrate=BITRATE) buffered = io.BufferedWriter(self._output) self._picam.start_recording(self._stream_encoder, FileOutput(buffered)) self._stop_event.clear() self._ready_event.clear() self._watch_thread = threading.Thread(target=self._watch_playlist, daemon=True) self._watch_thread.start() self.running = True logger.info("Camera started — waiting for first HLS segment") def _watch_playlist(self) -> None: """Signal ready once the m3u8 playlist exists and has at least one segment.""" playlist = HLS_DIR / "stream.m3u8" while not self._stop_event.is_set(): if playlist.exists(): content = playlist.read_text() if ".ts" in content: logger.info("HLS playlist ready") self._ready_event.set() return time.sleep(0.25) def wait_until_ready(self, timeout: float = 15.0) -> bool: if not PICAMERA_AVAILABLE: return True return self._ready_event.wait(timeout) def stop(self) -> None: if not self.running: return # stop any active recording first if self.recording: self.stop_recording() self._stop_event.set() self._ready_event.set() # unblock any waiters if self._picam: self._picam.stop_recording() self._picam.close() self._picam = None if self._output: self._output.close() if self._ffmpeg and self._ffmpeg.stdin: self._ffmpeg.stdin.close() if self._ffmpeg: try: self._ffmpeg.wait(timeout=5) except subprocess.TimeoutExpired: self._ffmpeg.kill() self._ffmpeg = None if HLS_DIR.exists(): shutil.rmtree(HLS_DIR) self.running = False logger.info("Camera stopped") # ── Recording ──────────────────────────────────────────────────────────── def start_recording(self) -> Path: """ Start recording to an MP4 file. Can run independently of or simultaneously with the HLS stream. Returns the output file path. """ with self._lock: if self.recording: raise RuntimeError("Already recording") RECORDINGS_DIR.mkdir(parents=True, exist_ok=True) timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") output_path = RECORDINGS_DIR / f"recording_{timestamp}.mp4" if not PICAMERA_AVAILABLE: # mock mode — create an empty placeholder file output_path.touch() self._current_recording_path = output_path self._recording_started_at = datetime.now(UTC) self.recording = True logger.info("Mock recording started: %s", output_path) return output_path if self._picam is None: # camera not yet streaming — start it temporarily in record-only mode self._picam = Picamera2() config = self._picam.create_video_configuration( main={"size": (1280, 720)}, ) self._picam.configure(config) self._picam.set_controls( { "Brightness": 0.1, "Contrast": 1.1, "Saturation": 1.1, "Sharpness": 1.0, "AwbEnable": True, "AeEnable": True, } ) # pipe raw H.264 → ffmpeg → MP4 container ffmpeg_cmd = [ "ffmpeg", "-loglevel", "warning", "-f", "h264", "-i", "pipe:0", "-c:v", "copy", "-movflags", "+faststart", str(output_path), ] self._record_ffmpeg = subprocess.Popen( ffmpeg_cmd, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) record_pipe = PipeOutput(self._record_ffmpeg) record_buffered = io.BufferedWriter(record_pipe) self._record_encoder = H264Encoder(bitrate=BITRATE) if self.running: # camera already running for streaming — add a second encoder output self._picam.start_encoder( self._record_encoder, FileOutput(record_buffered), ) else: # no stream active — start camera just for recording self._picam.start_recording( self._record_encoder, FileOutput(record_buffered), ) self._current_recording_path = output_path self._recording_started_at = datetime.now(UTC) self.recording = True logger.info("Recording started: %s", output_path) return output_path def stop_recording(self) -> Path | None: """Stop the active recording and finalise the MP4 file.""" with self._lock: if not self.recording: return None path = self._current_recording_path if PICAMERA_AVAILABLE and self._record_encoder is not None: if self.running: # streaming still active — only stop the recording encoder self._picam.stop_encoder(self._record_encoder) # type: ignore[union-attr] else: # recording-only mode — stop the whole camera if self._picam: self._picam.stop_recording() self._picam.close() self._picam = None self._record_encoder = None if self._record_ffmpeg: if self._record_ffmpeg.stdin: self._record_ffmpeg.stdin.close() try: self._record_ffmpeg.wait(timeout=10) except subprocess.TimeoutExpired: self._record_ffmpeg.kill() self._record_ffmpeg = None self.recording = False self._current_recording_path = None self._recording_started_at = None logger.info("Recording stopped: %s", path) return path def recording_status(self) -> dict[str, object]: """Return current recording state for the API.""" return { "recording": self.recording, "path": ( str(self._current_recording_path) if self._current_recording_path else None ), "started_at": ( self._recording_started_at.isoformat() if self._recording_started_at else None ), } def list_recordings(self) -> list[dict[str, object]]: """Return metadata for all saved recordings, newest first.""" if not RECORDINGS_DIR.exists(): return [] recordings = [] for f in sorted(RECORDINGS_DIR.glob("recording_*.mp4"), reverse=True): stat = f.stat() recordings.append( { "filename": f.name, "size_bytes": stat.st_size, "created_at": datetime.fromtimestamp( stat.st_ctime, tz=UTC ).isoformat(), "duration_seconds": None, # could be derived via ffprobe if needed } ) return recordings # ── Properties ─────────────────────────────────────────────────────────── @property def hls_dir(self) -> Path: return HLS_DIR camera = Camera()