slashtechno
/
wyzely-detect
mirror of https://github.com/slashtechno/wyzely-detect
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixed scaling 🎉

This commit is contained in:
slashtechno 2023-10-14 19:25:27 -05:00
parent 7d942ee456
commit 1d17bb629b
Signed by: slashtechno
GPG Key ID: 8EC1D9D9286C2B17
3 changed files with 71 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
set_detect_notify/__main__.py
View File

@ -197,7 +197,9 @@ def main():
# The following is stuff for people # The following is stuff for people
# This is still in the for loop as each result, no matter if anything is detected, will be present. # This is still in the for loop as each result, no matter if anything is detected, will be present.
# Thus, there will always be one result (r) # Thus, there will always be one result (r)
if face_details := utils.recognize_face(path_to_directory=Path(args.faces_directory), run_frame=run_frame): if face_details := utils.recognize_face(
path_to_directory=Path(args.faces_directory), run_frame=run_frame
):
plot_boxes.append(face_details) plot_boxes.append(face_details)
objects_and_peoples = notify.thing_detected( objects_and_peoples = notify.thing_detected(
thing_name=face_details["label"], thing_name=face_details["label"],
@ -209,12 +211,12 @@ def main():
ntfy_url=args.ntfy_url, ntfy_url=args.ntfy_url,
) )
# The following is stuff for objects # The following is stuff for objects
# Setup dictionary of object names # Setup dictionary of object names
if objects_and_peoples["objects"] == {} or objects_and_peoples["objects"] is None: if (
objects_and_peoples["objects"] == {}
or objects_and_peoples["objects"] is None
):
for name in r.names.values(): for name in r.names.values():
objects_and_peoples["objects"][name] = { objects_and_peoples["objects"][name] = {
"last_detection_time": None, "last_detection_time": None,
@ -274,8 +276,7 @@ def main():
ntfy_url=args.ntfy_url, ntfy_url=args.ntfy_url,
) )
# TODO: On 10-14-2023, while testing, it seemed the bounding box was too low. Troubleshoot if it's a plotting problem. # To debug plotting, use r.plot() to cross reference the bounding boxes drawn by the plot_label() and r.plot()
# To do so, use r.plot() to cross reference the bounding box drawn by the plot_label function and r.plot()
frame_to_show = utils.plot_label( frame_to_show = utils.plot_label(
boxes=plot_boxes, boxes=plot_boxes,
full_frame=frame, full_frame=frame,
@ -297,5 +298,6 @@ def main():
video_capture.release() video_capture.release()
cv2.destroyAllWindows() cv2.destroyAllWindows()
if __name__ == "__main__": if __name__ == "__main__":
main() main()

22
set_detect_notify/utils/notify.py
View File

@ -2,10 +2,9 @@ import httpx
import time import time
''' """
Structure of objects_and_peoples Structure of objects_and_peoples
Really, the only reason peoples is a separate dictionary is to prevent duplicates, though it just makes the code more complicated. Really, the only reason peoples is a separate dictionary is to prevent duplicates, though it just makes the code more complicated.
TODO: Make a function to check if a person is in the objects dictionary and vice versa
{ {
"objects": { "objects": {
"object_name": { "object_name": {
@ -22,7 +21,7 @@ TODO: Make a function to check if a person is in the objects dictionary and vice
}, },
}, },
} }
''' """
# objects_and_peoples = {} # objects_and_peoples = {}
@ -33,12 +32,12 @@ def thing_detected(
detection_window: int = 15, detection_window: int = 15,
detection_duration: int = 2, detection_duration: int = 2,
notification_window: int = 15, notification_window: int = 15,
ntfy_url: str = "https://ntfy.sh/set-detect-notify" ntfy_url: str = "https://ntfy.sh/set-detect-notify",
) -> dict: ) -> dict:
''' """
A function to make sure 2 seconds of detection is detected in 15 seconds, 15 seconds apart. A function to make sure 2 seconds of detection is detected in 15 seconds, 15 seconds apart.
Takes a dict that will be retured with the updated detection times. MAKE SURE TO SAVE THE RETURNED DICTIONARY Takes a dict that will be retured with the updated detection times. MAKE SURE TO SAVE THE RETURNED DICTIONARY
''' """
# "Alias" the objects and peoples dictionaries so it's easier to work with # "Alias" the objects and peoples dictionaries so it's easier to work with
respective_type = objects_and_peoples[detection_type] respective_type = objects_and_peoples[detection_type]
@ -93,22 +92,18 @@ def thing_detected(
# (re)send notification # (re)send notification
# Check if detection has been ongoing for 2 seconds or more in the past 15 seconds # Check if detection has been ongoing for 2 seconds or more in the past 15 seconds
if ( if (
respective_type[thing_name]["detection_duration"] respective_type[thing_name]["detection_duration"] >= detection_duration
>= detection_duration
and time.time() - respective_type[thing_name]["last_detection_time"] and time.time() - respective_type[thing_name]["last_detection_time"]
<= detection_window <= detection_window
): ):
# If the last notification was more than 15 seconds ago, then send a notification # If the last notification was more than 15 seconds ago, then send a notification
if ( if (
respective_type[thing_name]["last_notification_time"] is None respective_type[thing_name]["last_notification_time"] is None
or time.time() or time.time() - respective_type[thing_name]["last_notification_time"]
- respective_type[thing_name]["last_notification_time"]
> notification_window > notification_window
): ):
respective_type[thing_name]["last_notification_time"] = time.time() respective_type[thing_name]["last_notification_time"] = time.time()
print( print(f"Detected {thing_name} for {detection_duration} seconds")
f"Detected {thing_name} for {detection_duration} seconds"
)
headers = construct_ntfy_headers( headers = construct_ntfy_headers(
title=f"{thing_name} detected", title=f"{thing_name} detected",
tag="rotating_light", tag="rotating_light",
@ -140,4 +135,3 @@ def send_notification(data: str, headers: dict, url: str):
if url is None or data is None: if url is None or data is None:
raise ValueError("url and data cannot be None") raise ValueError("url and data cannot be None")
httpx.post(url, data=data.encode("utf-8"), headers=headers) httpx.post(url, data=data.encode("utf-8"), headers=headers)

46
set_detect_notify/utils/utils.py
View File

@ -5,6 +5,7 @@ from deepface import DeepFace
first_face_try = True first_face_try = True
def plot_label( def plot_label(
# list of dicts with each dict containing a label, x1, y1, x2, y2 # list of dicts with each dict containing a label, x1, y1, x2, y2
boxes: list = None, boxes: list = None,
@ -18,20 +19,23 @@ def plot_label(
view_scale: float = None, view_scale: float = None,
font: int = cv2.FONT_HERSHEY_SIMPLEX, font: int = cv2.FONT_HERSHEY_SIMPLEX,
): ):
# x1 and y1 are the top left corner of the box
# x2 and y2 are the bottom right corner of the box
# Example scaling: full_frame: 1 run_frame: 0.5 view_frame: 0.25
view_frame = cv2.resize(full_frame, (0, 0), fx=view_scale, fy=view_scale) view_frame = cv2.resize(full_frame, (0, 0), fx=view_scale, fy=view_scale)
for thing in boxes: for thing in boxes:
cv2.rectangle( cv2.rectangle(
# Image # Image
view_frame, view_frame,
# Start point # Top left corner
( (
int(thing["x1"] * (run_scale / view_scale)), int((thing["x1"] / run_scale) * view_scale),
int(thing["y1"] * (run_scale / view_scale)), int((thing["y1"] / run_scale) * view_scale),
), ),
# End point # Bottom right corner
( (
int(thing["x2"] * (run_scale / view_scale)), int((thing["x2"] / run_scale) * view_scale),
int(thing["y2"] * (run_scale / view_scale)), int((thing["y2"] / run_scale) * view_scale),
), ),
# Color # Color
(0, 255, 0), (0, 255, 0),
@ -45,8 +49,8 @@ def plot_label(
thing["label"], thing["label"],
# Origin # Origin
( (
int(thing["x1"] * (run_scale / view_scale)), int((thing["x1"] / run_scale) * view_scale),
int(thing["y1"] * (run_scale / view_scale)), int((thing["y1"] / run_scale) * view_scale) - 10,
), ),
# Font # Font
font, font,
@ -65,7 +69,7 @@ def recognize_face(
# opencv image # opencv image
run_frame: np.ndarray = None, run_frame: np.ndarray = None,
) -> np.ndarray: ) -> np.ndarray:
''' """
Accepts a path to a directory of images of faces to be used as a refference Accepts a path to a directory of images of faces to be used as a refference
In addition, accepts an opencv image to be used as the frame to be searched In addition, accepts an opencv image to be used as the frame to be searched
@ -85,7 +89,7 @@ def recognize_face(
Point is, `name` is the name of the person in the images in the directory `name` Point is, `name` is the name of the person in the images in the directory `name`
That name will be used as the label for the face in the frame That name will be used as the label for the face in the frame
''' """
global first_face_try global first_face_try
# If it's the first time the function is being run, remove representations_vgg_face.pkl, if it exists # If it's the first time the function is being run, remove representations_vgg_face.pkl, if it exists
@ -99,9 +103,17 @@ def recognize_face(
# face_dataframes is a vanilla list of dataframes # face_dataframes is a vanilla list of dataframes
try: try:
face_dataframes = DeepFace.find(run_frame, db_path=str(path_to_directory), enforce_detection=True, silent=True) face_dataframes = DeepFace.find(
run_frame,
db_path=str(path_to_directory),
enforce_detection=True,
silent=True,
)
except ValueError as e: except ValueError as e:
if str(e) == "Face could not be detected. Please confirm that the picture is a face photo or consider to set enforce_detection param to False.": if (
str(e)
== "Face could not be detected. Please confirm that the picture is a face photo or consider to set enforce_detection param to False."
):
return None return None
# Iteate over the dataframes # Iteate over the dataframes
for df in face_dataframes: for df in face_dataframes:
@ -120,14 +132,18 @@ def recognize_face(
"x2": df.iloc[-1]["source_x"] + df.iloc[-1]["source_w"], "x2": df.iloc[-1]["source_x"] + df.iloc[-1]["source_w"],
"y2": df.iloc[-1]["source_y"] + df.iloc[-1]["source_h"], "y2": df.iloc[-1]["source_y"] + df.iloc[-1]["source_h"],
} }
# After some brief testing, it seems positve matches are > 0.3
# I have not seen any false positives, so there is no threashold yet
distance = df.iloc[-1]["VGG-Face_cosine"] distance = df.iloc[-1]["VGG-Face_cosine"]
# if 0.5 < distance < 0.7: # if 0.5 < distance < 0.7:
# label = "Unknown" # label = "Unknown"
to_return = dict(label=label, **coordinates) to_return = dict(label=label, **coordinates)
print(f'Confindence: {distance}, filname: {path_to_image.name}, to_return: {to_return}') print(
f"Confindence: {distance}, filname: {path_to_image.name}, to_return: {to_return}"
)
return to_return return to_return
''' """
Example dataframe, for reference Example dataframe, for reference
identity (path to image) | source_x | source_y | source_w | source_h | VGG-Face_cosine (pretty much the confidence \_('_')_/) identity (path to image) | source_x | source_y | source_w | source_h | VGG-Face_cosine (pretty much the confidence \_('_')_/)
''' """