#!/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()