ashish/frigate
ashish
/
frigate
1
1
Fork 0
Code Issues 2 Pull Requests Packages Projects 1 Releases Wiki Activity

Added try/except to prevent exiting of sunset thread

This commit is contained in:
Ashish D'Souza 2023-07-30 02:43:47 -04:00
parent fa157287e1
commit 9d490f8ce4
1 changed files with 29 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
webcontrol/src/detection.py
View File

@ -3,7 +3,7 @@ import sys
import json
from time import sleep
from threading import Thread
from datetime import datetime, timedelta
from datetime import datetime, time, timedelta
import requests
from flask import Blueprint, request, jsonify
@ -14,32 +14,40 @@ blueprint = Blueprint('detection', __name__)
def get_sunset() -> datetime:
IPGEOLOCATION_API_KEY = os.environ['IPGEOLOCATION_API_KEY']
ipgeolocation_api_response = requests.get(f'https://api.ipgeolocation.io/astronomy?apiKey={IPGEOLOCATION_API_KEY}&location=Winter+Haven,+FL')
ipgeolocation_api_response.raise_for_status()
astronomical_json = json.loads(ipgeolocation_api_response.content)
sunset_time = datetime.strptime(astronomical_json['sunset'], '%H:%M').time()
sunset_date = datetime.now().date()
if sunset_time < datetime.now().time():
# Sunset has already passed today
sunset_date += timedelta(days=1)
return datetime.combine(sunset_date, sunset_time)
try:
IPGEOLOCATION_API_KEY = os.environ['IPGEOLOCATION_API_KEY']
ipgeolocation_api_response = requests.get(f'https://api.ipgeolocation.io/astronomy?apiKey={IPGEOLOCATION_API_KEY}&location=Winter+Haven,+FL')
ipgeolocation_api_response.raise_for_status()
astronomical_json = json.loads(ipgeolocation_api_response.content)
sunset_time = datetime.strptime(astronomical_json['sunset'], '%H:%M').time()
except Exception as e:
print(e, file=sys.stderr)
sunset_time = time(20, 00, 00)
finally:
if sunset_time < datetime.now().time():
# Sunset has already passed today
sunset_date += timedelta(days=1)
return datetime.combine(sunset_date, sunset_time)
def reset_all_camera_detection_at_sunset():
while True:
sunset = get_sunset()
print(f'Waiting until {sunset} to reset detection for all cameras...', file=sys.stderr)
seconds_until_sunset = (sunset - datetime.now()).total_seconds()
sleep(seconds_until_sunset)
try:
sunset = get_sunset()
print(f'Waiting until {sunset} to reset detection for all cameras...', file=sys.stderr)
seconds_until_sunset = (sunset - datetime.now()).total_seconds()
sleep(seconds_until_sunset)
# Get all camera names
frigate_api_response = requests.get('http://frigate:5000/api/config')
frigate_api_response.raise_for_status()
all_camera_names = json.loads(frigate_api_response.content)['cameras'].keys()
for camera_name in all_camera_names:
set_camera_detection(camera_name, True)
# Get all camera names
frigate_api_response = requests.get('http://frigate:5000/api/config')
frigate_api_response.raise_for_status()
all_camera_names = json.loads(frigate_api_response.content)['cameras'].keys()
for camera_name in all_camera_names:
set_camera_detection(camera_name, True)
except Exception as e:
print(e, file=sys.stderr)
def set_camera_detection(camera: str, value: bool, delay: int = 0) -> None: