A few weeks ago, I upgraded the dashcam in my truck. The one I had was just a little inexpensive one that my grandmother gave me for Christmas one year I think. It wasn’t anything fancy, wasn’t super high resolution, but it worked. It was a pain in the butt though if I ever wanted footage because it had no wireless connectivity at all. The only way to review the footage was to take the sd card out and copy the files to my computer.
Enter the BlackVue DR900S
This bad boy is cloud capable as well as direct WiFi. I can bring up the full 4K UHD video directly on my phone while near the car using the cam’s built in WiFi, or from my office desk connecting to the cloud (if you have a mobile hotspot in your vehicle).
I’ve been very happy with it so far, and today, I took it a step further. Using Home Assistant and a plugin called AppDaemon, I setup an automation that runs when ever the dashcam connects to my home WiFi network. When the dashcam goes from “away” to “home”, I fire an event that I subscribe to in AppDaemon. In that event handler, I connect to the camera, download any new videos that are on the camera to my home server. I then delete any old video’s off my home server that are over a year old (I know, I’m a pack rat).
At first, I wasn’t sure how easy this was going to be, but it turned out to be rather simple, and I knocked it out in a couple of hours.
Below you can find the code I used. Here is my HomeAssistant automation definition:
- alias: 'Truck Dashcam Connected'
trigger:
- entity_id: device_tracker.unifi_00_25_42_31_c8_a7_default
from: not_home
platform: state
to: home
action:
- event: truck_dashcam_connected
event_data: {}
AppDaemon app definition:
truckdashcamdownload:
module: blackvuedashcamdownload
class: BlackVueDashCamDownload
event_to_listen_for: truck_dashcam_connected
smb_server_name: !secret dashcam_server
smb_server_address: !secret dashcam_server_address
smb_server_domain: !secret dashcam_domain
smb_timeout: 15
smb_username: !secret dashcam_username
smb_password: !secret dashcam_password
smb_share: !secret dashcam_share
smb_path: "Truck/"
dashcam_ip: !secret truck_dashcam_ip
download_videosonly: False
rear_camera: True
days_to_keep: 365
AppDaemon module: (please forgive my Python coding, this is the 2nd piece of Python I’ve ever written :))
import appdaemon.plugins.hass.hassapi as hass
from smb.SMBConnection import SMBConnection
import requests
import datetime
import io
import locale
class BlackVueDashCamDownload(hass.Hass):
def initialize(self):
self.event_to_listen_for = self.args.get("event_to_listen_for", "dashcam_connected")
self.smb_server_name = self.args.get("smb_server_name")
self.smb_server_address = self.args.get("smb_server_address", self.smb_server_name)
self.smb_server_port = self.args.get("smb_server_port", 139)
self.smb_timeout = self.args.get("smb_timeout", 30)
self.smb_server_domain = self.args.get("smb_server_domain", "")
self.smb_use_ntlm_v2 = self.args.get("smb_use_ntlm_v2", True)
self.smb_sign_options = self.args.get("smb_sign_options", 2)
self.smb_is_direct_tcp = self.args.get("smb_is_direct_tcp", False)
self.smb_username = self.args.get("smb_username")
self.smb_password = self.args.get("smb_password")
self.smb_share = self.args.get("smb_share")
self.smb_path = self.args.get("smb_path", "/").replace("\\","/")
self.dashcam_ip = self.args.get("dashcam_ip")
self.download_videosonly = self.args.get("download_videosonly", True)
self.rear_camera = self.args.get("rear_camera", False)
self.days_to_keep = self.args.get("days_to_keep", 30)
self.listen_event(self.download_from_dashcam, self.event_to_listen_for)
self.log(f"Listening for {self.event_to_listen_for} event")
self.log("DashCamDownload.initialize() complete")
def download_from_dashcam(self, event, data, kwargs):
self.log(f"Starting DashCam Download for Camera: {self.dashcam_ip}")
try:
url = f"http://{self.dashcam_ip}/blackvue_vod.cgi"
r = requests.get(url)
files_on_camera = {}
existing_files = {}
files_to_download = []
files_downloaded = 0
bytes_to_download = 0
bytes_downloaded = 0
if r.status_code == 200:
content = r.text.replace("n:/Record/","").replace("F.mp4","").replace("R.mp4","").replace(",s:1000000","") # get a list of base filenames from the result
file_list = content.split("\r\n")
file_list.pop(0) # remove first line: "v:1.00"
for f in file_list:
filename = f.strip()
if filename == "":
continue
files_on_camera[f"{filename}F.mp4"] = True
if not self.download_videosonly:
files_on_camera[f"{filename}F.thm"] = True
if self.rear_camera:
files_on_camera[f"{filename}R.mp4"] = True
if not self.download_videosonly:
files_on_camera[f"{filename}R.thm"] = True
if not self.download_videosonly:
files_on_camera[f"{filename}.gps"] = True
files_on_camera[f"{filename}.3gf"] = True
conn = SMBConnection(self.smb_username, self.smb_password, "DASHCAMCOPY", self.smb_server_name, self.smb_server_domain, self.smb_use_ntlm_v2, self.smb_sign_options, self.smb_is_direct_tcp )
connected = conn.connect(self.smb_server_address, port = self.smb_server_port, timeout = self.smb_timeout)
if not connected:
self.error(f"Unable to connect to {self.smb_server_address}, port = {self.smb_server_port}")
return
self.log(f"Connected to //{self.smb_server_address}/{self.smb_share}")
results = conn.listPath(self.smb_share, self.smb_path)
self.log(f"Found {len(results)} existing files on //{self.smb_server_address}/{self.smb_share}/{self.smb_path}")
now = datetime.datetime.now()
epoch = datetime.datetime.utcfromtimestamp(0)
now_seconds = (now - epoch).total_seconds()
for r in results:
if not r.isDirectory:
age = ((now_seconds - r.last_write_time)/86400)
if age > self.days_to_keep and not r.filename in files_on_camera:
conn.deleteFiles(self.smb_share, f"{self.smb_path}/{r.filename}")
else:
existing_files[r.filename] = r.file_size
i = 0
total = len(files_on_camera)
for c in files_on_camera:
i = i + 1
percent = int((i / total) * 100)
url = f"http://{self.dashcam_ip}/Record/{c}"
download = True
size_on_camera = 0
if i % 100 == 0:
self.log(f"Preprocessing: {percent}% complete, {i} of {total}: {c}")
h = requests.head(url)
if h.status_code == 204:
download = False
if download:
if 'Content-Length' in h.headers:
size_on_camera = float(h.headers['Content-Length'])
if c in existing_files:
if size_on_camera != existing_files[c]:
self.log(f"Redownloading {c}, file size mismatch. On Camera: {size_on_camera}, On Disk: {existing_files[c]}")
else:
download = False
if download:
files_to_download.append(c)
bytes_to_download = bytes_to_download + size_on_camera
megabytes = bytes_to_download / 1024.0 / 1024.0
gigabytes = megabytes / 1024.0
megabytesstring = locale.format("%.0f", megabytes, grouping=True)
gigabytesstring = locale.format("%.4f", gigabytes, grouping=True)
self.log(f"need to download {len(files_to_download):n} files, {gigabytesstring} GB")
percent = 0
total = len(files_to_download)
i = 0
for f in files_to_download:
i = i + 1
url = f"http://{self.dashcam_ip}/Record/{f}"
d = requests.get(url, timeout = 3)
if d.status_code == 200:
conn.storeFile(self.smb_share, f"{self.smb_path}{f}", io.BytesIO(d.content))
self.log(f"Downloaded new DashCam file {url} to //{self.smb_server_address}/{self.smb_share}/{self.smb_path}{f}, {d.headers['Content-Length']} bytes")
bytes_downloaded = bytes_downloaded + float(d.headers['Content-Length'])
files_downloaded = files_downloaded + 1
else:
self.log(f"Unable to download DashCam file {url}, response code: {d.status_code}")
if bytes_to_download > 0:
percent = int((bytes_downloaded / bytes_to_download) * 100)
megabytes = bytes_downloaded / 1024.0 / 1024.0
bytesstring = locale.format("%.0f", megabytes, grouping=True)
self.log(f"Downloading: {percent}% complete, file {i} of {total}, {bytesstring} MB of {megabytesstring} MB completed")
megabytes = bytes_downloaded / 1024.0 / 1024.0
bytesstring = locale.format("%.0f", megabytes, grouping=True)
self.log(f"Dashcam Completely Copied Successfully, {files_downloaded:n} files, {bytesstring} MB downloaded")
finally:
if conn:
conn.close()
self.log("Complete")
And there you have it. I have a modular piece of code that I can reuse for additional cameras. After doing this, I’ve decided I’m going to get a cam for my wife, and when my stepson is old enough to drive, he’ll have one as well 🙂 All I have to do is add another application definition to my AppDaemon apps.yaml file like this:
vandashcamdownload:
module: blackvuedashcamdownload
class: BlackVueDashCamDownload
event_to_listen_for: van_dashcam_connected
smb_server_name: !secret dashcam_server
smb_server_address: !secret dashcam_server_address
smb_server_domain: !secret dashcam_domain
smb_timeout: 15
smb_username: !secret dashcam_username
smb_password: !secret dashcam_password
smb_share: !secret dashcam_share
smb_path: "Van/"
dashcam_ip: !secret van_dashcam_ip
download_videosonly: False
rear_camera: True
days_to_keep: 365
And now, I have two automatic dashcam downloads. The files goto a share on my server:
You can then use the BlackVue software on your PC to view the footage with audio/gps/speed all available like below:
2 comments on “Home Assistant + BlackVue Dashcam: Automatically archiving all footage”
Hey, just wanted to say thanks for the write up. I was just poking around the web to see if it was possible to connect my Garmin Dashcam to HA and I found your page. I’m thinking I’d like to get a live feed from the camera when it’s on my Wifi and use it like an additional security cam, your post here will help get me started. Cheers!
Thank you! After some fiddling I got it this work. The main problem was actually getting AppDaemon work with imports.