#!/usr/bin/env python3 from datetime import datetime from pathlib import Path import os import signal import sys import time import numpy as np from libcamera import controls from picamera2 import Picamera2 from picamera2.encoders import H264Encoder from picamera2.outputs import CircularOutput # --------------------------------------------------------------------------- # Recording configuration # --------------------------------------------------------------------------- OUTPUT_DIRECTORY = Path("/mnt/output/camera") VIDEO_SIZE = (1280, 720) ANALYSIS_SIZE = (320, 240) FRAME_RATE = 15 BITRATE = 2_000_000 # Continue recording for this long after the last detected movement. QUIET_SECONDS = 10.0 # Include this much video from immediately before movement was detected. PRE_RECORD_SECONDS = 2.0 # Allow auto-exposure and white balance to settle before arming detection. WARMUP_SECONDS = 3.0 # --------------------------------------------------------------------------- # Motion-detection configuration # --------------------------------------------------------------------------- # A pixel must change by at least this much, from 0 to 255, to count as changed. PIXEL_CHANGE_THRESHOLD = 25 # Motion is detected when this fraction of analysed pixels has changed. # 0.015 means 1.5 percent of the image. MOTION_AREA_THRESHOLD = 0.015 # Analyse every second pixel horizontally and vertically to reduce CPU use. ANALYSIS_SAMPLE_STEP = 2 shutdown_requested = False def log(message: str) -> None: """Print a timestamped console message immediately.""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {message}", flush=True) def request_shutdown(_signum: int, _frame: object) -> None: global shutdown_requested shutdown_requested = True def main() -> int: OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) if not os.access(OUTPUT_DIRECTORY, os.W_OK): log(f"ERROR: Output directory is not writable: {OUTPUT_DIRECTORY}") return 1 signal.signal(signal.SIGINT, request_shutdown) signal.signal(signal.SIGTERM, request_shutdown) camera = Picamera2() configuration = camera.create_video_configuration( main={ "size": VIDEO_SIZE, "format": "YUV420", }, lores={ "size": ANALYSIS_SIZE, "format": "YUV420", }, controls={ "FrameRate": FRAME_RATE, }, buffer_count=6, ) camera.configure(configuration) # Camera Module 3 continuously adjusts focus when necessary. camera.set_controls({ "AfMode": controls.AfModeEnum.Continuous, }) encoder = H264Encoder( bitrate=BITRATE, repeat=True, iperiod=FRAME_RATE, ) prebuffer_frames = max( 1, round(FRAME_RATE * PRE_RECORD_SECONDS), ) circular_output = CircularOutput( buffersize=prebuffer_frames, ) encoder.output = circular_output recording = False recording_path: Path | None = None recording_trigger_time = 0.0 last_motion_time = 0.0 camera_started = False encoder_started = False try: log("Starting camera") camera.start() camera_started = True # The encoder runs continuously. While idle, CircularOutput keeps only # the configured short pre-trigger buffer in RAM. camera.start_encoder(encoder) encoder_started = True log( f"Camera running at {VIDEO_SIZE[0]}x{VIDEO_SIZE[1]}, " f"{FRAME_RATE} FPS" ) log( f"Warming up for {WARMUP_SECONDS:.1f} seconds before " "arming motion detection" ) time.sleep(WARMUP_SECONDS) analysis_width, analysis_height = ANALYSIS_SIZE first_frame = camera.capture_array("lores") previous_frame = first_frame[:analysis_height, :analysis_width] previous_frame = previous_frame[ ::ANALYSIS_SAMPLE_STEP, ::ANALYSIS_SAMPLE_STEP, ].astype(np.int16) log("Motion detection armed") log( f"Recordings will be saved in {OUTPUT_DIRECTORY}; " f"recording stops after {QUIET_SECONDS:.0f} seconds " "without motion" ) while not shutdown_requested: frame = camera.capture_array("lores") # YUV420 starts with the full-resolution greyscale Y plane. current_frame = frame[:analysis_height, :analysis_width] current_frame = current_frame[ ::ANALYSIS_SAMPLE_STEP, ::ANALYSIS_SAMPLE_STEP, ].astype(np.int16) difference = np.abs(current_frame - previous_frame) changed_pixels = np.count_nonzero( difference >= PIXEL_CHANGE_THRESHOLD ) changed_fraction = changed_pixels / difference.size motion_detected = ( changed_fraction >= MOTION_AREA_THRESHOLD ) now = time.monotonic() if motion_detected: last_motion_time = now if not recording: filename_timestamp = datetime.now().strftime( "%Y-%m-%d_%H-%M-%S" ) recording_path = ( OUTPUT_DIRECTORY / f"{filename_timestamp}.h264" ) # Start writing the circular pre-buffer and all subsequent # encoded H.264 packets directly to the file. circular_output.fileoutput = str(recording_path) circular_output.start() recording = True recording_trigger_time = now log( "RECORDING STARTED | " f"motion area={changed_fraction * 100:.2f}%" ) log(f"Output file: {recording_path}") elif ( recording and now - last_motion_time >= QUIET_SECONDS ): circular_output.stop() recording = False elapsed = now - recording_trigger_time if ( recording_path is not None and recording_path.exists() ): size_mib = ( recording_path.stat().st_size / (1024 * 1024) ) log( "RECORDING STOPPED | " f"no motion for {QUIET_SECONDS:.0f} seconds | " f"trigger-to-stop duration={elapsed:.1f} seconds | " f"size={size_mib:.2f} MiB" ) log(f"Saved: {recording_path}") else: log( "RECORDING STOPPED | " f"no motion for {QUIET_SECONDS:.0f} seconds" ) recording_path = None previous_frame = current_frame except KeyboardInterrupt: log("Shutdown requested") except Exception as error: log(f"ERROR: {type(error).__name__}: {error}") return 1 finally: if recording: try: circular_output.stop() if ( recording_path is not None and recording_path.exists() ): size_mib = ( recording_path.stat().st_size / (1024 * 1024) ) log( "RECORDING STOPPED | " f"program shutting down | " f"size={size_mib:.2f} MiB" ) log(f"Saved: {recording_path}") except Exception as error: log(f"Error closing recording: {error}") if encoder_started: try: camera.stop_encoder() except Exception: pass if camera_started: try: camera.stop() except Exception: pass camera.close() log("Camera stopped") return 0 if __name__ == "__main__": raise SystemExit(main())