diff --git a/pyproject.toml b/pyproject.toml index 7934293..4f2498f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,11 +1,11 @@ [tool.poetry] -name = "set-detect-notify" +name = "set_detect_notify" version = "0.1.0" description = "Detect all the things" authors = ["slashtechno <77907286+slashtechno@users.noreply.github.com>"] license = "MIT" readme = "README.md" -packages = [{include = "set-detect-notify"}] +packages = [{include = "set_detect_notify"}] [tool.poetry.dependencies] # python = "^3.10" @@ -41,3 +41,6 @@ build-backend = "poetry.core.masonry.api" # Where possible, `black` will attempt to format to 88 characters # However, setting ruff to 135 will allow for longer lines that can't be auto-formatted line-length = 135 + +[tool.poetry.scripts] +set-detect-notify = "set_detect_notify.__main__:main" \ No newline at end of file diff --git a/set-detect-notify/__init__.py b/set_detect_notify/__init__.py similarity index 100% rename from set-detect-notify/__init__.py rename to set_detect_notify/__init__.py diff --git a/set-detect-notify/__main__.py b/set_detect_notify/__main__.py similarity index 99% rename from set-detect-notify/__main__.py rename to set_detect_notify/__main__.py index 43b3510..549c5ce 100644 --- a/set-detect-notify/__main__.py +++ b/set_detect_notify/__main__.py @@ -275,5 +275,5 @@ def main(): video_capture.release() cv2.destroyAllWindows() - -main() +if __name__ == "__main__": + main() diff --git a/set-detect-notify/utils/__init__.py b/set_detect_notify/utils/__init__.py similarity index 100% rename from set-detect-notify/utils/__init__.py rename to set_detect_notify/utils/__init__.py diff --git a/set-detect-notify/utils/notify.py b/set_detect_notify/utils/notify.py similarity index 97% rename from set-detect-notify/utils/notify.py rename to set_detect_notify/utils/notify.py index 780ef5d..f3a37aa 100644 --- a/set-detect-notify/utils/notify.py +++ b/set_detect_notify/utils/notify.py @@ -1,134 +1,134 @@ -import httpx -import time - - -''' -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. -TODO: Make a function to check if a person is in the objects dictionary and vice versa -{ - "objects": { - "object_name": { - "last_detection_time": float, - "detection_duration": float, - "last_notification_time": float, - }, - }, - "peoples": { - "person_name": { - "last_detection_time": float, - "detection_duration": float, - "last_notification_time": float, - }, - }, -} -''' -# objects_and_peoples = {} - - -def thing_detected( - thing_name: str, - objects_and_peoples: dict, - detection_type: str = "objects", - detection_window: int = 15, - detection_duration: int = 2, - notification_window: int = 15, - ntfy_url: str = "https://ntfy.sh/set-detect-notify" - ) -> dict: - ''' - 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 - ''' - - # "Alias" the objects and peoples dictionaries so it's easier to work with - respective_type = objects_and_peoples[detection_type] - - # (re)start cycle - if ( - # If the object has not been detected before - respective_type[thing_name]["last_detection_time"] is None - # If the last detection was more than 15 seconds ago - or time.time() - respective_type[thing_name]["last_detection_time"] - > detection_window - ): - # Set the last detection time to now - respective_type[thing_name]["last_detection_time"] = time.time() - print(f"First detection of {thing_name} in this detection window") - # This line is important. It resets the detection duration when the object hasn't been detected for a while - # If detection duration is None, don't print anything. - # Otherwise, print that the detection duration is being reset due to inactivity - if respective_type[thing_name]["detection_duration"] is not None: - print( - f"Resetting detection duration for {thing_name} since it hasn't been detected for {detection_window} seconds" # noqa: E501 - ) - respective_type[thing_name]["detection_duration"] = 0 - else: - # Check if the last notification was less than 15 seconds ago - # If it was, then don't do anything - if ( - time.time() - respective_type[thing_name]["last_detection_time"] - <= notification_window - ): - pass - # If it was more than 15 seconds ago, reset the detection duration - # This effectively resets the notification timer - else: - print("Notification timer has expired - resetting") - respective_type[thing_name]["detection_duration"] = 0 - respective_type[thing_name]["detection_duration"] += ( - time.time() - respective_type[thing_name]["last_detection_time"] - ) - # print("Updating detection duration") - respective_type[thing_name]["last_detection_time"] = time.time() - - # (re)send notification - # Check if detection has been ongoing for 2 seconds or more in the past 15 seconds - if ( - respective_type[thing_name]["detection_duration"] - >= detection_duration - and time.time() - respective_type[thing_name]["last_detection_time"] - <= detection_window - ): - # If the last notification was more than 15 seconds ago, then send a notification - if ( - respective_type[thing_name]["last_notification_time"] is None - or time.time() - - respective_type[thing_name]["last_notification_time"] - > notification_window - ): - respective_type[thing_name]["last_notification_time"] = time.time() - print( - f"Detected {thing_name} for {detection_duration} seconds" - ) - headers = construct_ntfy_headers( - title=f"{thing_name} detected", - tag="rotating_light", - priority="default", - ) - send_notification( - data=f"{thing_name} detected for {detection_duration} seconds", - headers=headers, - url=ntfy_url, - ) - # Reset the detection duration - print("Just sent a notification - resetting detection duration") - respective_type[thing_name]["detection_duration"] = 0 - - # Take the aliased objects_and_peoples and update the respective dictionary - objects_and_peoples[detection_type] = respective_type - return objects_and_peoples - - -def construct_ntfy_headers( - title: str = "Object/Person Detected", - tag="rotating_light", # https://docs.ntfy.sh/publish/#tags-emojis - priority="default", # https://docs.ntfy.sh/publish/#message-priority -) -> dict: - return {"Title": title, "Priority": priority, "Tags": tag} - - -def send_notification(data: str, headers: dict, url: str): - if url is None or data is None: - raise ValueError("url and data cannot be None") - httpx.post(url, data=data.encode("utf-8"), headers=headers) - +import httpx +import time + + +''' +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. +TODO: Make a function to check if a person is in the objects dictionary and vice versa +{ + "objects": { + "object_name": { + "last_detection_time": float, + "detection_duration": float, + "last_notification_time": float, + }, + }, + "peoples": { + "person_name": { + "last_detection_time": float, + "detection_duration": float, + "last_notification_time": float, + }, + }, +} +''' +# objects_and_peoples = {} + + +def thing_detected( + thing_name: str, + objects_and_peoples: dict, + detection_type: str = "objects", + detection_window: int = 15, + detection_duration: int = 2, + notification_window: int = 15, + ntfy_url: str = "https://ntfy.sh/set-detect-notify" + ) -> dict: + ''' + 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 + ''' + + # "Alias" the objects and peoples dictionaries so it's easier to work with + respective_type = objects_and_peoples[detection_type] + + # (re)start cycle + if ( + # If the object has not been detected before + respective_type[thing_name]["last_detection_time"] is None + # If the last detection was more than 15 seconds ago + or time.time() - respective_type[thing_name]["last_detection_time"] + > detection_window + ): + # Set the last detection time to now + respective_type[thing_name]["last_detection_time"] = time.time() + print(f"First detection of {thing_name} in this detection window") + # This line is important. It resets the detection duration when the object hasn't been detected for a while + # If detection duration is None, don't print anything. + # Otherwise, print that the detection duration is being reset due to inactivity + if respective_type[thing_name]["detection_duration"] is not None: + print( + f"Resetting detection duration for {thing_name} since it hasn't been detected for {detection_window} seconds" # noqa: E501 + ) + respective_type[thing_name]["detection_duration"] = 0 + else: + # Check if the last notification was less than 15 seconds ago + # If it was, then don't do anything + if ( + time.time() - respective_type[thing_name]["last_detection_time"] + <= notification_window + ): + pass + # If it was more than 15 seconds ago, reset the detection duration + # This effectively resets the notification timer + else: + print("Notification timer has expired - resetting") + respective_type[thing_name]["detection_duration"] = 0 + respective_type[thing_name]["detection_duration"] += ( + time.time() - respective_type[thing_name]["last_detection_time"] + ) + # print("Updating detection duration") + respective_type[thing_name]["last_detection_time"] = time.time() + + # (re)send notification + # Check if detection has been ongoing for 2 seconds or more in the past 15 seconds + if ( + respective_type[thing_name]["detection_duration"] + >= detection_duration + and time.time() - respective_type[thing_name]["last_detection_time"] + <= detection_window + ): + # If the last notification was more than 15 seconds ago, then send a notification + if ( + respective_type[thing_name]["last_notification_time"] is None + or time.time() + - respective_type[thing_name]["last_notification_time"] + > notification_window + ): + respective_type[thing_name]["last_notification_time"] = time.time() + print( + f"Detected {thing_name} for {detection_duration} seconds" + ) + headers = construct_ntfy_headers( + title=f"{thing_name} detected", + tag="rotating_light", + priority="default", + ) + send_notification( + data=f"{thing_name} detected for {detection_duration} seconds", + headers=headers, + url=ntfy_url, + ) + # Reset the detection duration + print("Just sent a notification - resetting detection duration") + respective_type[thing_name]["detection_duration"] = 0 + + # Take the aliased objects_and_peoples and update the respective dictionary + objects_and_peoples[detection_type] = respective_type + return objects_and_peoples + + +def construct_ntfy_headers( + title: str = "Object/Person Detected", + tag="rotating_light", # https://docs.ntfy.sh/publish/#tags-emojis + priority="default", # https://docs.ntfy.sh/publish/#message-priority +) -> dict: + return {"Title": title, "Priority": priority, "Tags": tag} + + +def send_notification(data: str, headers: dict, url: str): + if url is None or data is None: + raise ValueError("url and data cannot be None") + httpx.post(url, data=data.encode("utf-8"), headers=headers) + diff --git a/set-detect-notify/utils/utils.py b/set_detect_notify/utils/utils.py similarity index 100% rename from set-detect-notify/utils/utils.py rename to set_detect_notify/utils/utils.py