553 lines
21 KiB
Python
Executable File
553 lines
21 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import os
|
|
import shutil
|
|
import argparse
|
|
import picamera
|
|
from led import LED
|
|
from switch import Switch
|
|
from time import time, sleep
|
|
from threading import Thread, Lock
|
|
from random import randbytes
|
|
from movement import Adxl345Spi
|
|
from math import fabs
|
|
|
|
|
|
def get_usb_storage_device(desired_device=None):
|
|
with open("/proc/partitions") as file:
|
|
file.readline() #skip heading line
|
|
file.readline() #skip empty line
|
|
device_info = [
|
|
[ item.strip() for item in line.strip().split() ]
|
|
for line in file
|
|
]
|
|
devices = [
|
|
device[3]
|
|
for device in device_info
|
|
if len(device) >= 3 and device[0] in ("8", "259")
|
|
]
|
|
primary_devices = [
|
|
device[3]
|
|
for device in device_info
|
|
if len(device) >= 3 and device[0] in ("8", "259") and (int(device[1]) % 16) == 0
|
|
]
|
|
|
|
block_class_path = "/sys/class/block"
|
|
|
|
primary_usb_devices = [
|
|
device
|
|
for device in primary_devices
|
|
if (
|
|
os.path.islink(f"{block_class_path}/{device}") and
|
|
os.path.realpath(f"{block_class_path}/{device}").find("/usb") > 0
|
|
)
|
|
]
|
|
|
|
usb_partitions = sorted(
|
|
[
|
|
partition
|
|
for partition in devices
|
|
for usb_device in primary_usb_devices
|
|
if partition.startswith(usb_device) and usb_device != partition
|
|
],
|
|
reverse=True #assuming the last partition of the last is correct
|
|
)
|
|
|
|
if desired_device is not None:
|
|
if desired_device in usb_partitions:
|
|
return desired_device
|
|
return None
|
|
return usb_partitions[0]
|
|
|
|
def mount_usb_device(device, id):
|
|
mnt_path = f"/mnt/{id}"
|
|
if not os.path.ismount(mnt_path):
|
|
os.system(f"mkdir -p {mnt_path}")
|
|
os.system(f"mount {device} {mnt_path}")
|
|
return mnt_path
|
|
|
|
|
|
|
|
class Dashcam():
|
|
def __init__(
|
|
self, sequence_count=10, sequence_length=60, resolution=(1920, 1080),
|
|
video_type="h264", video_name_prefix="video-dashcam", bitrate = 17000000,
|
|
framerate=30, video_file_path="/opt/dashcam", pin_btn_pwr=11, pin_btn_cpy=12,
|
|
pin_btn_stop=13, pin_btn_info=15, pin_led_cpy=29, pin_led_pwr=33,
|
|
pin_led_info=37, led_pwr_dim_perc=5, g_force_limit=1.5, salt_bytes=4):
|
|
self.pin_btn_cpy = pin_btn_cpy
|
|
self.pin_btn_pwr = pin_btn_pwr
|
|
self.pin_btn_info = pin_btn_info
|
|
self.pin_btn_stop = pin_btn_stop
|
|
self.pin_led_pwr = pin_led_pwr
|
|
self.pin_led_cpy = pin_led_cpy
|
|
self.pin_led_info = pin_led_info
|
|
self.pin_led_pwr_dim_percent = led_pwr_dim_perc
|
|
self.pin_blink_seconds = 2
|
|
self.pin_blink_on_seconds = 0.1
|
|
self.g_force_limit = g_force_limit
|
|
|
|
self.video_sequence_seconds = sequence_length
|
|
self.video_sequence_count = sequence_count
|
|
self.video_resolution = resolution
|
|
self.video_name_prefix = video_name_prefix
|
|
self.video_file_path=video_file_path
|
|
self.video_file_path_legal=f"{self.video_file_path}/legal"
|
|
self.video_type = video_type if video_type in ("h264", "mjpeg") else "h264"
|
|
self.video_bit_rate = bitrate
|
|
self.video_frame_rate = framerate
|
|
|
|
# using a salt to not eventually overwrite files
|
|
# after an unexpected reboot in car; is like
|
|
# a unique identifier for an ongoing record session
|
|
self.video_name_salt = randbytes(salt_bytes).hex()
|
|
|
|
self.LED_data = LED(self.pin_led_cpy)
|
|
self.LED_power = LED(self.pin_led_pwr)
|
|
self.LED_info = LED(self.pin_led_info)
|
|
|
|
|
|
self.BTN_data = Switch(self.pin_btn_cpy)
|
|
self.BTN_power = Switch(self.pin_btn_pwr)
|
|
self.BTN_stop = Switch(self.pin_btn_stop)
|
|
self.BTN_info = Switch(self.pin_btn_info)
|
|
|
|
self.file_lock = Lock()
|
|
self.camera_lock = Lock()
|
|
|
|
self.adxl345 = Adxl345Spi()
|
|
|
|
self.camera = picamera.PiCamera(
|
|
resolution=self.video_resolution,
|
|
framerate=self.video_frame_rate
|
|
)
|
|
self.camera_state = 0 #0: off, 1: turndown, 2: on
|
|
self.info_led_state = 0
|
|
self.segment_ctr = 0
|
|
self.video_filename = ""
|
|
|
|
def get_video_id(self):
|
|
return self.video_name_salt
|
|
|
|
def set_video_path(self, path):
|
|
self.video_file_path = path
|
|
self.video_file_path_legal = f"{self.video_file_path}/legal"
|
|
|
|
def __del__(self):
|
|
del self.LED_data, self.LED_power
|
|
|
|
def _dashcam_video_thread(self):
|
|
self.video_filename = (
|
|
f"{self.video_name_prefix}_"
|
|
f"{int(time())}-{self.video_name_salt}-"
|
|
f"{self.segment_ctr}.{self.video_type}"
|
|
)
|
|
video_path = f"{self.video_file_path}/{self.video_filename}"
|
|
print(f"Recording to '{video_path}'.")
|
|
self.camera.start_recording(
|
|
video_path, format=self.video_type, bitrate=self.video_bit_rate
|
|
)
|
|
self.camera.wait_recording(self.video_sequence_seconds)
|
|
|
|
while self.camera_state > 1:
|
|
self.segment_ctr += 1
|
|
tmp_video_filename = (
|
|
f"{self.video_name_prefix}_"
|
|
f"{int(time())}-{self.video_name_salt}-"
|
|
f"{self.segment_ctr}.{self.video_type}"
|
|
)
|
|
video_path = f"{self.video_file_path}/{tmp_video_filename}"
|
|
print(f"Recording to '{video_path}'.")
|
|
self.camera.split_recording(video_path)
|
|
# as the copy thread callback might be a bit too fast,
|
|
# we manage to set the final new filename AFTER the switch
|
|
# which guarantees, that the file is really finished.
|
|
self.video_filename = tmp_video_filename
|
|
self.camera.wait_recording(self.video_sequence_seconds)
|
|
self.camera.stop_recording()
|
|
self.camera_state = 0
|
|
|
|
|
|
def _dashcam_file_cleanup_thread(self):
|
|
while True:
|
|
self.file_lock.acquire()
|
|
|
|
video_file_list = self.get_directory_file_list(
|
|
self.video_file_path, self.video_type
|
|
)
|
|
video_file_list_legal = self.get_video_file_list_legal(
|
|
video_file_list, buffer=1
|
|
)
|
|
|
|
delete_video_file_list = [
|
|
video_file
|
|
for video_file in video_file_list
|
|
if video_file not in video_file_list_legal
|
|
]
|
|
|
|
for del_video_file in delete_video_file_list:
|
|
print(f"DELETE file '{del_video_file}'")
|
|
try:
|
|
os.remove(f"{self.video_file_path}/{del_video_file}")
|
|
except FileNotFoundError:
|
|
print(
|
|
f"WARNING! File '{self.video_file_path}/{del_video_file}'"
|
|
" is gone. Ignoreing file. Continue"
|
|
)
|
|
|
|
self.file_lock.release()
|
|
sleep(self.video_sequence_seconds)
|
|
|
|
def _led_power_heartbeat(self, LED):
|
|
LED.set_duty_cycle(self.pin_led_pwr_dim_percent)
|
|
sleep(0.15)
|
|
LED.set_duty_cycle(0)
|
|
sleep(0.15)
|
|
LED.set_duty_cycle(self.pin_led_pwr_dim_percent)
|
|
sleep(0.2)
|
|
LED.set_duty_cycle(0)
|
|
|
|
def _dashcam_powerled_thread(self):
|
|
round_cntr = 0
|
|
round_time = 0.5
|
|
|
|
LED_is_on = True
|
|
|
|
LED_state_switch = {
|
|
0: self.LED_info,
|
|
2: self.LED_power
|
|
}
|
|
|
|
while True:
|
|
if self.camera_state in LED_state_switch:
|
|
LED = LED_state_switch[self.camera_state]
|
|
if (self.info_led_state % 4) == 0 and (round_cntr % 60) == 0:
|
|
self._led_power_heartbeat(LED)
|
|
elif (self.info_led_state % 4) == 1 and (round_cntr % 1) == 0:
|
|
self._led_power_heartbeat(LED)
|
|
elif (self.info_led_state % 4) == 2:
|
|
if not LED_is_on:
|
|
LED.set_on()
|
|
LED.set_duty_cycle(self.pin_led_pwr_dim_percent)
|
|
LED_is_on = True
|
|
else:
|
|
if LED_is_on:
|
|
LED.set_off()
|
|
LED_is_on = False
|
|
elif self.camera_state == 1:
|
|
if not LED_is_on:
|
|
self.LED_info.set_duty_cycle(self.pin_led_pwr_dim_percent)
|
|
LED_is_on = True
|
|
else:
|
|
self.LED_info.set_off()
|
|
LED_is_on = False
|
|
#sleep to make catch changes in e.g. info led status changes
|
|
sleep(round_time)
|
|
round_cntr += round_time
|
|
|
|
def _g_force_surveillance(self):
|
|
for _ in range(5):
|
|
#cleanup at every start/coldstart (like at car ;) )
|
|
x,y,z = self.adxl345.get_acceleration()
|
|
sleep(1)
|
|
while self.camera_state > 0:
|
|
accl_xyz = self.adxl345.get_acceleration()
|
|
if any(
|
|
fabs(val) > self.g_force_limit
|
|
for val in accl_xyz
|
|
):
|
|
self.save_video_file_legal(self.LED_data)
|
|
sleep(0.5)
|
|
self.adxl345.stop()
|
|
|
|
def get_directory_file_list(self, path, filetype):
|
|
return [
|
|
file
|
|
for file in os.listdir(path)
|
|
if (
|
|
os.path.isfile(os.path.join(path,file)) and
|
|
file.endswith(f'.{filetype}')
|
|
)
|
|
]
|
|
|
|
def get_video_file_list_legal(self, video_file_list, reverse=True, buffer=0):
|
|
prefix_match_sorted_reduced_video_fileid_list = sorted(
|
|
[
|
|
file.removeprefix(
|
|
f"{self.video_name_prefix}_"
|
|
).removesuffix(
|
|
f'.{self.video_type}'
|
|
)
|
|
for file in video_file_list
|
|
if file.startswith(f"{self.video_name_prefix}_")
|
|
], key = (
|
|
lambda x: (
|
|
f"{int(x.split('-')[1] == self.video_name_salt)}{x}"
|
|
)
|
|
), reverse=reverse
|
|
)
|
|
|
|
return [
|
|
f"{self.video_name_prefix}_{fileid}.{self.video_type}"
|
|
for fileid in prefix_match_sorted_reduced_video_fileid_list
|
|
][:self.video_sequence_count+buffer]
|
|
|
|
def save_video_file_legal(self, LED):
|
|
self.file_lock.acquire()
|
|
LED.set_on()
|
|
|
|
video_file_list_legal = self.get_video_file_list_legal(
|
|
self.get_directory_file_list(
|
|
self.video_file_path, self.video_type
|
|
), reverse=False
|
|
)
|
|
|
|
timestamp = int(time())
|
|
legal_path = f"{self.video_file_path_legal}/{timestamp}_utc"
|
|
os.makedirs(legal_path, exist_ok=True)
|
|
|
|
current_video = self.video_filename
|
|
is_active_saving = current_video in video_file_list_legal
|
|
if is_active_saving:
|
|
video_file_list_legal.remove(current_video)
|
|
|
|
for video_file in video_file_list_legal:
|
|
src = f"{self.video_file_path}/{video_file}"
|
|
dst = f"{legal_path}/INCIDENT_{video_file}"
|
|
print(f"Copy '{src}' to '{dst}'.")
|
|
try:
|
|
shutil.copyfile(src, dst)
|
|
except FileNotFoundError:
|
|
print(f"WARNING! File '{src}' is gone. Ignoreing file. Continue")
|
|
while is_active_saving and self.video_filename == current_video:
|
|
#waiting for current video to finish
|
|
sleep(5)
|
|
if is_active_saving:
|
|
src = f"{self.video_file_path}/{current_video}"
|
|
dst = f"{legal_path}/INCIDENT_{current_video}"
|
|
print(f"Copy '{src}' to '{dst}'.")
|
|
try:
|
|
shutil.copyfile(src, dst)
|
|
except FileNotFoundError:
|
|
print(f"WARNING! File '{src}' is gone. Ignoreing file. Continue")
|
|
|
|
print("Copy done.")
|
|
|
|
for round in range(int(self.pin_blink_seconds / self.pin_blink_on_seconds)):
|
|
if round % 2 == 0:
|
|
LED.set_on()
|
|
else:
|
|
LED.set_off()
|
|
sleep(self.pin_blink_on_seconds)
|
|
|
|
LED.set_off()
|
|
self.file_lock.release()
|
|
|
|
def _button_copy_functor(self, input):
|
|
if input == 0:
|
|
self.save_video_file_legal(self.LED_data)
|
|
|
|
def _button_start_functor(self, input):
|
|
if input == 0:
|
|
if self.camera_state == 0:
|
|
self.camera_lock.acquire()
|
|
self.camera_state = 2
|
|
self.video_thread = Thread(target=self._dashcam_video_thread)
|
|
self.g_force_thread = Thread(target=self._g_force_surveillance)
|
|
self.video_thread.start()
|
|
self.LED_power.set_duty_cycle(self.pin_led_pwr_dim_percent)
|
|
self.g_force_thread.start()
|
|
sleep(10) #mainly user notification via LED on
|
|
self.camera_lock.release()
|
|
|
|
def _button_stop_functor(self, input):
|
|
if input == 0:
|
|
if self.camera_state == 2:
|
|
self.camera_lock.acquire()
|
|
self.camera_state = 1
|
|
self.LED_power.set_off()
|
|
self.video_thread.join()
|
|
self.g_force_thread.join()
|
|
del self.video_thread
|
|
del self.g_force_thread
|
|
self.camera_lock.release()
|
|
|
|
def _button_info_functor(self, input):
|
|
if input == 0:
|
|
# in order to keep numbers small and as we probably won't
|
|
# have more than 100 blinking states, lets keep it 0 < x < 100 !
|
|
self.info_led_state = (self.info_led_state + 1) % 100
|
|
|
|
def do_warning(self):
|
|
# just blink at info LED!
|
|
# can be used when e.g. mountpoint is unavailable!!!
|
|
for _ in range(10):
|
|
self.LED_info.set_on()
|
|
sleep(0.5)
|
|
self.LED_info.set_off()
|
|
sleep(0.5)
|
|
|
|
|
|
def start(self):
|
|
os.makedirs(self.video_file_path, exist_ok=True)
|
|
os.makedirs(self.video_file_path_legal, exist_ok=True)
|
|
|
|
self.power_led_thread = Thread(target=self._dashcam_powerled_thread)
|
|
self.clean_thread = Thread(target=self._dashcam_file_cleanup_thread)
|
|
|
|
self.BTN_data.set_functor(self._button_copy_functor)
|
|
self.BTN_power.set_functor(self._button_start_functor)
|
|
self.BTN_stop.set_functor(self._button_stop_functor)
|
|
self.BTN_info.set_functor(self._button_info_functor)
|
|
|
|
self.power_led_thread.start()
|
|
self.clean_thread.start()
|
|
|
|
self._button_start_functor(0)
|
|
|
|
def join_clean_thread(self):
|
|
self.clean_thread.join()
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="""DashCam-app following the German traffic and GDPR (DSGVO)
|
|
laws. Is designed to work on any Raspberry PI that provides PiCamera
|
|
library in Python; this fits basically for any 32-Bit driven RPI.
|
|
Mainly used in combination with a RPI Zero W v1.1, which is totally
|
|
sufficient for this purpose.
|
|
Instead of a single, increasing video file, the input video stream is
|
|
splitted into e.g. 60s video chunks; only the last e.g. 10 chunks are kept.
|
|
This allows to have e.g. the last 10 minutes stored but also regularly cleaned
|
|
in order to stay conform towards GDPR.
|
|
There is a special incident procedure: when an attached button is pressed,
|
|
the last e.g. 10 legal video chunks are separately stored and not cleaned
|
|
to e.g. provide some legal information in any case of accident/emergency.
|
|
In addition, two LEDs can be attached to print Power/Heartbeat information
|
|
and to indicate the current, active data copy in any case of accident (this
|
|
will only happen, when you press the button).
|
|
"""
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-s", "--video_chunk_duration", metavar="S", type=int, required=False,
|
|
default=60, help="Length of a single stored video chunk."
|
|
)
|
|
parser.add_argument(
|
|
"-c", "--video_chunk_count", metavar="C", type=int, required=False,
|
|
default=10, help=(
|
|
"Max. number of sequential video chunks that are stored in parallel"
|
|
"on the disk; this correlates do GDPR laws."
|
|
)
|
|
)
|
|
parser.add_argument(
|
|
"-p", "--video_file_path", metavar="P", type=str, required=False,
|
|
default="/opt/dashcam", help="Location to store dashcam video chunks"
|
|
)
|
|
parser.add_argument(
|
|
"-f", "--video_file_prefix", metavar="F", type=str, required=False,
|
|
default="video-dashcam", help=(
|
|
"Filename prefix for the locally stored video file chunks"
|
|
)
|
|
)
|
|
parser.add_argument(
|
|
"-r", "--video_resolution", metavar="R", nargs=2, type=int, required=False,
|
|
default=(1920, 1080), help="Length of a single stored video chunk."
|
|
)
|
|
parser.add_argument(
|
|
"-br", "--video_bitrate", metavar="BR", type=int, required=False,
|
|
default=7000000, help="Bitrate used to store videos."
|
|
)
|
|
parser.add_argument(
|
|
"-vf", "--video_format", metavar="VF", type=str, required=False,
|
|
default="h264", choices=("h264","mjpeg"),help="File format used to store videos."
|
|
)
|
|
parser.add_argument(
|
|
"-lc", "--pin_led_copy", metavar="PLC", type=int, required=False,
|
|
default=29, help="Pin number in GPIO.BOARD layout for a data-copy LED."
|
|
)
|
|
parser.add_argument(
|
|
"-lp", "--pin_led_power", metavar="PLP", type=int, required=False,
|
|
default=33, help="Pin number in GPIO.BOARD layout for a power LED."
|
|
)
|
|
parser.add_argument(
|
|
"-li", "--pin_led_info", metavar="PLP", type=int, required=False,
|
|
default=37, help="Pin number in GPIO.BOARD layout for an info LED."
|
|
)
|
|
parser.add_argument(
|
|
"-bp", "--pin_button_power", metavar="PBP", type=int, required=False,
|
|
default=11, help="Pin number in GPIO.BOARD layout for a start button."
|
|
)
|
|
parser.add_argument(
|
|
"-bc", "--pin_button_copy", metavar="PBC", type=int, required=False,
|
|
default=12, help="Pin number in GPIO.BOARD layout for a data copy button."
|
|
)
|
|
parser.add_argument(
|
|
"-bs", "--pin_button_stop", metavar="PBS", type=int, required=False,
|
|
default=13, help="Pin number in GPIO.BOARD layout for a stop button."
|
|
)
|
|
parser.add_argument(
|
|
"-bi", "--pin_button_info", metavar="PBI", type=int, required=False,
|
|
default=15, help="Pin number in GPIO.BOARD layout for a info-led control button."
|
|
)
|
|
parser.add_argument(
|
|
"-d", "--pin_power_dim_percent", metavar="LPD", type=int, required=False,
|
|
default=5, help="Percent dim for the Power LED; if it might be to bright."
|
|
)
|
|
parser.add_argument(
|
|
"-g", "--g_force_limit", metavar="G", type=int, required=False, default=1.5,
|
|
help="Threshold in terms of g-Force, when a data copy should be triggered."
|
|
)
|
|
parser.add_argument(
|
|
"--external_usb_storage_device", metavar="DEVICE", type=str, required=False,
|
|
help=(
|
|
"Dev shortcut, e.g. sda1, for a partition/device you want to use as "
|
|
"video storage. Will be directly written on it."
|
|
)
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
video_chunk_duration = args.video_chunk_duration
|
|
video_chunk_count = args.video_chunk_count
|
|
video_file_path = args.video_file_path
|
|
video_file_prefix = args.video_file_prefix
|
|
video_resolution = args.video_resolution
|
|
video_bitrate = args.video_bitrate
|
|
video_format = args.video_format
|
|
pin_button_copy = args.pin_button_copy
|
|
pin_button_power = args.pin_button_power
|
|
pin_button_stop = args.pin_button_stop
|
|
pin_button_info = args.pin_button_info
|
|
pin_led_power = args.pin_led_power
|
|
pin_led_copy = args.pin_led_copy
|
|
pin_led_info = args.pin_led_info
|
|
pin_power_dim_percent = args.pin_power_dim_percent
|
|
g_force_limit = args.g_force_limit
|
|
usb_storage = args.external_usb_storage_device if hasattr(args,'external_usb_storage_device') else None
|
|
|
|
|
|
dashcam = Dashcam(
|
|
sequence_count=video_chunk_count, sequence_length=video_chunk_duration,
|
|
resolution=video_resolution, video_type=video_format, bitrate=video_bitrate,
|
|
video_name_prefix=video_file_prefix, video_file_path=video_file_path,
|
|
pin_btn_cpy=pin_button_copy, pin_btn_pwr=pin_button_power,
|
|
pin_btn_info=pin_button_info, pin_btn_stop=pin_button_stop,
|
|
pin_led_cpy=pin_led_copy, pin_led_pwr=pin_led_power, pin_led_info=pin_led_info,
|
|
led_pwr_dim_perc=pin_power_dim_percent, g_force_limit=g_force_limit
|
|
)
|
|
|
|
|
|
if usb_storage is not None:
|
|
usb_device = get_usb_storage_device(usb_storage)
|
|
if usb_device is not None:
|
|
mount_path = mount_usb_device(f"/dev/{usb_device}", "dashcam-videodata")
|
|
dashcam.set_video_path(mount_path)
|
|
else:
|
|
dashcam.do_warning()
|
|
|
|
dashcam.start()
|
|
dashcam.join_clean_thread()
|
|
|
|
|
|
if __name__=='__main__':
|
|
main()
|