h3cx/plotcctv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Hector van der Aa
2026-07-29 13:29:46 +01:00
commit dcc44b05bd
8 changed files with 396 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.python-version Normal file
View File

@@ -0,0 +1 @@
3.13

0
README.md Normal file
View File

27
disk_connected_test.py Normal file
View File

@@ -0,0 +1,27 @@
import pyudev
def main() -> None:
context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by(subsystem="block")
print("Waiting for disks...")
for device in iter(monitor.poll, None):
if device.action == "add" and device.device_type == "disk":
print("\nDisk connected")
print(f"Device: {device.device_node}")
print(f"Name: {device.sys_name}")
print(f"Model: {device.get('ID_MODEL', 'Unknown')}")
print(f"Serial: {device.get('ID_SERIAL_SHORT', 'Unknown')}")
print(f"Bus: {device.get('ID_BUS', 'Unknown')}")
elif device.action == "remove" and device.device_type == "disk":
print("\nDisk disconnected")
print(f"Name: {device.sys_name}")
if __name__ == "__main__":
main()

11
led_test.py Normal file
View File

@@ -0,0 +1,11 @@
from gpiozero import LED
from time import sleep
red = LED(17)
yellow = LED(27)
blue = LED(22)
for led in (red, yellow, blue):
led.on()
sleep(1)
led.off()

27
main.py Normal file
View File

@@ -0,0 +1,27 @@
import pyudev
def main() -> None:
context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by(subsystem="block")
print("Waiting for disks...")
for device in iter(monitor.poll, None):
if device.action == "add" and device.device_type == "disk":
print("\nDisk connected")
print(f"Device: {device.device_node}")
print(f"Name: {device.sys_name}")
print(f"Model: {device.get('ID_MODEL', 'Unknown')}")
print(f"Serial: {device.get('ID_SERIAL_SHORT', 'Unknown')}")
print(f"Bus: {device.get('ID_BUS', 'Unknown')}")
elif device.action == "remove" and device.device_type == "disk":
print("\nDisk disconnected")
print(f"Name: {device.sys_name}")
if __name__ == "__main__":
main()

298
motion_detect.py Executable file
View File

@@ -0,0 +1,298 @@
#!/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())

9
pyproject.toml Normal file
View File

@@ -0,0 +1,9 @@
[project]
name = "plotcctv"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"pyudev>=0.24.4",
]

23
uv.lock generated Normal file
View File

@@ -0,0 +1,23 @@
version = 1
revision = 3
requires-python = ">=3.13"
[[package]]
name = "plotcctv"
version = "0.1.0"
source = { virtual = "." }
dependencies = [
{ name = "pyudev" },
]
[package.metadata]
requires-dist = [{ name = "pyudev", specifier = ">=0.24.4" }]
[[package]]
name = "pyudev"
version = "0.24.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/5e/1d/8bdbf651de1002e8b58fbe817bee22b1e8bfcdd24341d42c3238ce9a75f4/pyudev-0.24.4.tar.gz", hash = "sha256:e788bb983700b1a84efc2e88862b0a51af2a995d5b86bc9997546505cf7b36bc", size = 56135, upload-time = "2025-10-08T17:26:58.661Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/51/3dc0cd6498b24dea3cdeaed648568e3ca7454d41334d840b114156d7479f/pyudev-0.24.4-py3-none-any.whl", hash = "sha256:b3b6b01c68e6fc628428cc45ff3fe6c277afbb5d96507f14473ddb4a6b959e00", size = 62784, upload-time = "2025-10-08T17:26:57.664Z" },
]