299 lines
8.3 KiB
Python
Executable File
299 lines
8.3 KiB
Python
Executable File
#!/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())
|