shoofle/mirror-to-yesterday
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add support for he selector switch

Browse Source
This commit is contained in:
Shoofle 2024-08-04 11:25:12 -04:00
parent 5a77966102
commit 03c3f1a9f1
1 changed files with 99 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
webcam.py
View File

@ -1,7 +1,9 @@
from datetime import datetime, date, time, timedelta from datetime import datetime, date, time, timedelta
import os import os
import cv2 import cv2
from wakepy import keep import gpiod
from gpiod import Direction, Value
#from wakepy import keep
def five_minute_segment_from(the_time): def five_minute_segment_from(the_time):
# this is mocked out too be a one minute segment instead # this is mocked out too be a one minute segment instead
@ -25,85 +27,127 @@ if not camera.isOpened():
# video writer # video writer
fourcc = cv2.VideoWriter_fourcc(*'H264') fourcc = cv2.VideoWriter_fourcc(*'H264')
camera_fps = camera.get(cv2.CAP_PROP_FPS)
video_writer = cv2.VideoWriter(path_to_video_at(last_time_stamp), video_writer = cv2.VideoWriter(path_to_video_at(last_time_stamp),
fourcc, fourcc,
camera.get(cv2.CAP_PROP_FPS), camera_fps,
(int(camera.get(3)),int(camera.get(4))), (int(camera.get(3)),int(camera.get(4))), #resolution
(cv2.VIDEOWRITER_PROP_HW_ACCELERATION, cv2.VIDEO_ACCELERATION_ANY) (cv2.VIDEOWRITER_PROP_HW_ACCELERATION, cv2.VIDEO_ACCELERATION_ANY)
) )
# video feed from previously captured video # video feed from previously captured video
old_time = cv2.VideoCapture(path_to_video_at(last_time_stamp))
twenty_four_hours_ago = cv2.VideoCapture(path_to_video_at(last_time_stamp - timedelta(hours=24)))
twelve_hours_ago = cv2.VideoCapture(path_to_video_at(last_time_stamp - timedelta(hours=12)))
five_minutes_ago = cv2.VideoCapture(path_to_video_at(last_time_stamp - timedelta(minutes=5)))
# fullscreen window yess good # fullscreen window yess good
cv2.namedWindow("mirror", cv2.WINDOW_NORMAL) cv2.namedWindow("mirror", cv2.WINDOW_NORMAL)
cv2.setWindowProperty("mirror", cv2.WND_PROP_FULLSCREEN, cv2.WINDOW_FULLSCREEN) cv2.setWindowProperty("mirror", cv2.WND_PROP_FULLSCREEN, cv2.WINDOW_FULLSCREEN)
queue = [(False, None)] * 100 #queue of recent frames queue = [(False, None)] * math.floor(camera_fps * 32) #queue of 32 seconds of frames
idx = 0 # current index into the frame queue idx = 0 # current index into the frame queue
short_term_delay_frames = 20 # hoow many frames back are we going in the buffer when we're on a short term delay?
longterm = False # are we shoowing a long term delay? (versus a sshort term delay) lines = {
"five_second": 1,
"thirty_second": 2,
"five_minute": 3,
"twelve_hour": 4,
"twenty_four_hour": 5
}
#with keep.presenting(): #with keep.presenting():
with gpiod.request_lines(
"/dev/gpiochip0",
consumer="mirror-to-yesterday",
config={
lines.five_second: gpiod.LineSettings(direction=Direction.INPUT),
lines.thirty_second: gpiod.LineSettings(direction=Direction.INPUT),
lines.five_minute: gpiod.LineSettings(direction=Direction.INPUT),
lines.twelve_hour: gpiod.LineSettings(direction=Direction.INPUT),
lines.twenty_four_hour: gpiod.LineSettings(direction=Direction.INPUT)
}
) as io_lines:
while(True): while(True):
if last_time_stamp != five_minute_segment_from(datetime.now()): if last_time_stamp != five_minute_segment_from(datetime.now()):
# find file older than our long delay # find file older than our long delay
old_file = path_to_video_at(five_minute_segment_from(datetime.now() - long_delay - timedelta(minutes=1))) oldest_file_path = path_to_video_at(five_minute_segment_from(datetime.now() - timedelta(hours=24) - timedelta(minutes=10)))
if os.path.isfile(old_file): if os.path.isfile(oldest_file_path):
print(f"deleting old file {old_file}") print(f"deleting old file {oldest_file_path}")
os.remove(old_file) os.remove(oldest_file_path)
video_writer.release() video_writer.release()
newly_recording_file = path_to_video_at(five_minute_segment_from(datetime.now())) newly_recording_file_path = path_to_video_at(five_minute_segment_from(datetime.now()))
print(f"opening {newly_recording_file} too record this upcoming ssegment to") print(f"opening {newly_recording_file_path} too record this upcoming ssegment to")
video_writer = cv2.VideoWriter(newly_recording_file, video_writer = cv2.VideoWriter(newly_recording_file_path,
fourcc, fourcc,
camera.get(cv2.CAP_PROP_FPS), camera_fps,
(int(camera.get(3)), int(camera.get(4))), (int(camera.get(3)), int(camera.get(4))),
(cv2.VIDEOWRITER_PROP_HW_ACCELERATION, cv2.VIDEO_ACCELERATION_ANY) (cv2.VIDEOWRITER_PROP_HW_ACCELERATION, cv2.VIDEO_ACCELERATION_ANY)
) )
if twenty_four_hours_ago.isOpened():
if old_time.isOpened(): twenty_four_hours_ago.release()
old_time.release() twenty_four_hours_ago = cv2.VideoCapture(path_to_video_at(five_minute_segment_from(datetime.now() - timedelta(hours=24))))
if twelve_hours_ago.isOpened():
new_file = path_to_video_at(five_minute_segment_from(datetime.now() - long_delay)) twelve_hours_ago.release()
print(f"opening the previously recorded file {new_file} to read from for long mirror") twelve_hours_ago = cv2.VideoCapture(path_to_video_at(five_minute_segment_from(datetime.now() - timedelta(hours=12))))
old_time = cv2.VideoCapture(new_file, cv2.CAP_FFMPEG) if five_minutes_ago.isOpened():
five_minutes_ago.release()
five_minutes_ago = cv2.VideoCapture(path_to_video_at(five_minute_segment_from(datetime.now() - timedelta(minutes=5))))
last_time_stamp = five_minute_segment_from(datetime.now()) last_time_stamp = five_minute_segment_from(datetime.now())
ret, frame = camera.read() ret, frame = camera.read()
frame = cv2.flip(frame, 1) frame = cv2.flip(frame, 1)
# record the frame to the appropriate file, and also to the short term buffer # record the frame to the appropriate file, and also to the short term buffer
video_writer.write(frame) video_writer.write(frame)
queue[idx] = ret, frame queue[idx] = ret, frame
prev_frame_valid, prev_frame = queue[(idx - short_term_delay_frames) % len(queue)] frames = {
old_frame_valid, old_frame = old_time.read() "five_second": queue[(idx - math.floor(5*camera_fps)) % len(queue)],
if not longterm and prev_frame_valid: "thirty_second": queue[(idx - math.floor(30*camera_fps)) % len(queue)],
cv2.imshow('mirror', prev_frame) "five_minute": five_minutes_ago.read(),
elif longterm and old_frame_valid: "twelve_hour": twelve_hours_ago.read(),
cv2.imshow('mirror', old_frame) "twenty_four_hour": twenty_four_hours_ago.read(),
}
#valid, the_frame = frames.five_second
if io_lines.get_value(lines.twenty_four_hour):
valid, the_frame = frames.twenty_four_hour
elif io_lines.get_value(lines.twelve_hour):
valid, the_frame = frames.twelve_hour
elif io_lines.get_value(lines.five_minute):
valid, the_frame = frames.five_minute
elif io_lines.get_value(lines.thirty_second):
valid, the_frame = frames.thirty_second
else: #if io_lines.get_value(lines.five_second)
valid, the_frame = frames.five_second
if valid:
cv2.imshow("mirror", the_frame)
idx = (idx + 1) % len(queue)
key = cv2.waitKey(16) # listen for keyprress for 16 ms key = cv2.waitKey(16) # listen for keyprress for 16 ms
if key & 0xFF == ord('q') or key == 27: # q or esc if key & 0xFF == ord('q') or key == 27: # q or esc
break break
if key & 0xFF == ord('a'): """if key & 0xFF == ord('a'):
short_term_delay_frames += 1 short_term_delay_frames += 1
if key & 0xFF == ord('d'): if key & 0xFF == ord('d'):
short_term_delay_frames -= 1 short_term_delay_frames -= 1
if key & 0xFF == ord('z'): if key & 0xFF == ord('z'):
longterm = not longterm longterm = not longterm"""
idx = (idx + 1) % len(queue)
# After the loop release the cap object # After the loop release the cap object
camera.release() camera.release()
old_time.release()
video_writer.release() video_writer.release()
twenty_four_hours_ago.release()
twelve_hours_ago.release()
five_minutes_ago.release()
# Destroy all the windows # Destroy all the windows
cv2.destroyAllWindows() cv2.destroyAllWindows()