From ddae50341076f604d48f9d274447db5fae3d9839 Mon Sep 17 00:00:00 2001 From: Eli Date: Sat, 22 Feb 2025 17:13:21 -0500 Subject: [PATCH] fresh commit --- .gitignore | 3 + code.py | 94 ++++++++++ readme.md | 39 ++++ requirements.txt | 26 +++ squirrel_annoyer.py | 447 ++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 609 insertions(+) create mode 100644 .gitignore create mode 100644 code.py create mode 100644 readme.md create mode 100644 requirements.txt create mode 100644 squirrel_annoyer.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ee18991 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +venv/ +*.pyc +__pycache__/ diff --git a/code.py b/code.py new file mode 100644 index 0000000..1784180 --- /dev/null +++ b/code.py @@ -0,0 +1,94 @@ +import os +import time +import board +import ipaddress +import wifi +import socketpool +import pwmio +from time import sleep +from adafruit_minimqtt.adafruit_minimqtt import MQTT +from adafruit_motorkit import MotorKit + +print("Connecting to WiFi") +wifi.radio.connect(os.getenv('CIRCUITPY_WIFI_SSID'), os.getenv('CIRCUITPY_WIFI_PASSWORD')) +print("Connected to WiFi") + +pool = socketpool.SocketPool(wifi.radio) +print("My MAC addr:", [hex(i) for i in wifi.radio.mac_address]) +print("My IP address is", wifi.radio.ipv4_address) +ipv4 = ipaddress.ip_address("8.8.4.4") +print("Ping google.com: %f ms" % (wifi.radio.ping(ipv4)*1000)) + +# MQTT broker details +MQTT_BROKER = os.getenv('MQTT_BROKER') +MQTT_PORT = int(os.getenv('MQTT_PORT', 1883)) +MQTT_TOPIC = os.getenv('MQTT_TOPIC', 'squirrel/speaker') +MQTT_USERNAME = os.getenv('MQTT_USERNAME') +MQTT_PASSWORD = os.getenv('MQTT_PASSWORD') + +# Initialize the speaker pin +SPEAKER_PIN = board.A0 + +def play_tone(frequency, duration): + speaker = pwmio.PWMOut(SPEAKER_PIN, frequency=frequency, duty_cycle=0x4000) + sleep(duration) + speaker.deinit() + +def activate_speaker(): + print("Activating speaker!") + play_tone(440, 0.5) # A4 + sleep(0.1) + play_tone(880, 0.5) # A5 + sleep(0.1) + play_tone(660, 0.5) # E5 + +# Callback for incoming MQTT messages +def message_callback(client, topic, message): + print(f"Received message on topic {topic}: {message}") + # Existing behavior for speaker activation + if topic == MQTT_TOPIC and message == "fire": + activate_speaker() + # New behavior for heater/bedroom motor control + elif topic == "heater/bedroom": + if message == "on": + print("Heater on command received: rotating motor forward") + kit.motor1.throttle = 1 + time.sleep(1) # adjust duration as needed + kit.motor1.throttle = 0 + elif message == "off": + print("Heater off command received: rotating motor in reverse") + kit.motor1.throttle = -1 + time.sleep(1) + kit.motor1.throttle = 0 + +mqtt_client = MQTT( + broker=MQTT_BROKER, + port=MQTT_PORT, + username=MQTT_USERNAME, + password=MQTT_PASSWORD, + socket_pool=pool, + is_ssl=False, +) + +mqtt_client.on_message = message_callback + +print("Connecting to MQTT broker...") +try: + mqtt_client.connect() + print("Connected to MQTT broker!") +except Exception as e: + print(f"Failed to connect to MQTT broker: {e}") + +# Subscribe to both topics +mqtt_client.subscribe(MQTT_TOPIC) +mqtt_client.subscribe("heater/bedroom") + +print("Listening for messages...") +while True: + try: + mqtt_client.loop() + time.sleep(0.1) + except Exception as e: + print(f"An error occurred: {e}") + time.sleep(5) + diff --git a/readme.md b/readme.md new file mode 100644 index 0000000..cf55c94 --- /dev/null +++ b/readme.md @@ -0,0 +1,39 @@ +# Squirrel Annoyer Widget Thingy +I have a few birdfeeders, and enjoy supporting the local bird population. The birds are polite, taking a seed or two and then fluttering away. Squirrels, however, have a tendency to hop on the feeder and sit there for an hour at a time just chompity-chomping away. They scare away all the birds and eat an unfair share. + +## The Journey So Far +I should have started documenting this earlier, but so far I've probably tried 50 different solutions to keep the squirrels off the feeders without setting traps or laying poison or anything else quite so drastic. I actually am fond of my furry friends, and simply wish to enforce good manners. + +I've tried all the feeders you've seen advertised on TV. They spin, they use springs to cover the inlets, they practically wash the dishes and make the bed. Alas, no luck via feeder variation. I next tried my own physical modifications. I've got the big inverted wok-style hat on one of the feeders now. Like the previous attempts, it worked well for a week or two but eventually an intrepid tree-rat figured out just the right bit of acrobatics to slide over the edge and catch the feeder below. + +I tried electrifying the fence. I managed to shock *myself* several times, but the squirrels escaped unscathed. It turns out it's harder than expected to electrify something hanging in the middle of the air. A real electrician would probably have had it working in 3 minutes, but alas... + +I put barbed wire and razors between the feeders and where the squirrels come from. I felt kind of bad about it, until I watched them traipse past the traps with the grace of a Russian ballerina. + +I share a few of the highlights, but the journey had many more bumps and turns that I may write about later. + +## Fast-forward to Today +I've been enjoying dipping my toes into the hobby electronics world. I have Home Assistant set up and this has allowed me many opportunities to develop my tinkering abilities with cheap and affordable sensors and chips. I figured I might apply the same skills to my years-long battle against the small, grey nutty ninjas. + +I first set out to create a script that alerts me when a squirrel is on the feeder. You need a clear detection mechanism/trigger before anything else can be implemented. This logic is seen in `squirrel_annoyer.py`. I pull images from my cameras, and then use AI image processing to check for a squirrel. If we get a hit, an alarm sounds. Currently, the alarm is a rather alarming `scream.wav` I pulled from an copyright-free library online. My wife hates it. It scares the bajeesus out of me! But not just me... + +## squirrel\_annoyer.py +After getting the basic script working I used it "manually". I'd hear the alarm and rush outside to scare the squirrels off the feeder. Funnily enough, it seemed some classical conditioning started to work. The squirrels would hear the scream and jump off the feeder on their own (at least sometimes). + +For some technical notes, I have a lot of troubleshooting messages in there, and it should be somewhat self-explanatory. I used AI to generate the skeleton and then tweaked it from there. + +In the future, I want to use my library of positive/negatively-identified photos to train my own local AI which will save me the outrageous $10/month it currently costs using OpenAI's 4o model. + +## code.py +Realizing there might be something to the sound angle of attack, I next bought an ESP32-C6 from Adafruit and started working on a speaker setup that I could power with a battery. My idea is to play an obnoxiously loud sound that's high enough to be above human thresholds so I don't annoy the neighbors (just their dogs). + +The current iteration uses an MQTT trigger (sent by the previously mentioned script) and plays the sound via an cheap attached tweeter speaker (designed for higher frequency work). I've got a Lithium Ion battery attached to keep it going, and basic testing has showed the 6600 mAh battery lasts over a day without any current power saving features enabled or utilized. As the squirrels only come out during daylight I can charge every night. + +## Future Work +There's lots to do. I need to find a way to package up the speaker-widget and attach it to the speaker to take advantage of the inverse-square law and maximum amplitude/volume. This will need to be waterproof and fairly durable. Gee, I wish I had a 3D printer! + +Beyond that, I need to do some testing to get the right exact sound to play. The script on-device currently plays a short melody that I can hear for testing purposes, but eventually I don't want to hear it at all. + +I also need to train my own local image-detection model to save money on API calls. I plan to let the API run a while longer to build a really good library of yes/no's, and then it should be fairly quick work to my own model. + +And if all of this doesn't fit my final needs...then it's time to move on to attempt #52. Maybe a servo that bangs on the feeder? Pops a balloon full of water on the squirrel? Ideas are welcome. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b938f54 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,26 @@ +annotated-types==0.7.0 +anyio==4.8.0 +astral==3.2 +certifi==2025.1.31 +charset-normalizer==3.4.1 +colorama==0.4.6 +distro==1.9.0 +h11==0.14.0 +httpcore==1.0.7 +httpx==0.28.1 +idna==3.10 +jiter==0.8.2 +numpy==2.2.2 +openai==1.61.1 +opencv-python==4.11.0.86 +paho-mqtt==2.1.0 +pydantic==2.10.6 +pydantic_core==2.27.2 +pygame==2.6.1 +python-dotenv==1.0.1 +pytz==2025.1 +requests==2.32.3 +sniffio==1.3.1 +tqdm==4.67.1 +typing_extensions==4.12.2 +urllib3==2.3.0 diff --git a/squirrel_annoyer.py b/squirrel_annoyer.py new file mode 100644 index 0000000..6349104 --- /dev/null +++ b/squirrel_annoyer.py @@ -0,0 +1,447 @@ +# run by calling `nuts` on terminal. +# Alias configured in C:\Users\PC\OneDrive\Documents\WindowsPowerShell\Microsoft.PowerShell_profile.ps1 +# view aliases with `$profile` +import os +import time +import datetime +import logging +import cv2 +import base64 +import requests +import urllib3 +import pygame +import numpy as np +import random +import shutil # Import this to move files +import paho.mqtt.client as mqtt +from dotenv import load_dotenv +from openai import OpenAI +from colorama import Fore, Back, Style, init +from astral import LocationInfo +from astral.sun import sun +from datetime import datetime, timedelta +from pytz import timezone + +# ========================== +# Configuration +# ========================== +load_dotenv("/home/eli/git/scripts/squirrel_anoyer/.env") +CAMERA_IP = os.getenv("CAMERA_IP") +CAMERA_USER = os.getenv("CAMERA_USER") +CAMERA_PASS = os.getenv("CAMERA_PASS") +CAPTURE_INTERVAL = 60 # seconds between captures +DATA_FOLDER = "C:\\Users\\PC\\squirreldata" +IMAGES_FOLDER = os.path.join(DATA_FOLDER, "images") +POSITIVE_DETECTION_FOLDER = os.path.join(IMAGES_FOLDER, "positive_detection") +LOG_FILE = os.path.join(DATA_FOLDER, "log.txt") + +# MQTT Configuration +MQTT_BROKER = os.getenv("MQTT_BROKER") +MQTT_PORT = int(os.getenv("MQTT_PORT")) +MQTT_TOPIC = os.getenv("MQTT_TOPIC") +MQTT_USERNAME = os.getenv("MQTT_USERNAME") +MQTT_PASSWORD = os.getenv("MQTT_PASSWORD") + +iteration = 0 +urllib3.disable_warnings() # disable HTTPS cert warnings. See:https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings + +# Set to True for verbose print statements +DEBUG_DEFAULT = False + +# Crop coordinates (left, top, right, bottom) +CROP_COORDS = (1700, 375, 2000, 812) + +# ========================== +# Setup +# ========================== +os.makedirs(POSITIVE_DETECTION_FOLDER, exist_ok=True) +os.makedirs(IMAGES_FOLDER, exist_ok=True) + +logging.basicConfig(filename=LOG_FILE, + format='%(asctime)s %(levelname)s: %(message)s', + level=logging.INFO) +# Initialize colorama +init(autoreset=True) + +# ========================== +# Helper Functions +# ========================== + +def debug_print(msg, debug=DEBUG_DEFAULT, style="normal"): + """ + Prints debug messages with styled formatting based on the style parameter. + Supported styles: highlight, danger, warn, muted, whimsylicious + """ + if not debug: + return + + # Define styles + if style == "highlight": + formatted_msg = Fore.GREEN + Style.BRIGHT + msg + elif style == "danger": + formatted_msg = Fore.RED + Style.BRIGHT + msg + elif style == "warn": + formatted_msg = Fore.YELLOW + Style.BRIGHT + msg + elif style == "muted": + formatted_msg = Fore.WHITE + Style.DIM + msg + elif style == "whimsylicious": + # Generate a random mix of colors for each character + formatted_msg = "".join( + random.choice([ + Fore.RED, Fore.GREEN, Fore.YELLOW, Fore.BLUE, + Fore.MAGENTA, Fore.CYAN, Back.RED, Back.GREEN, + Back.YELLOW, Back.BLUE, Back.MAGENTA, Back.CYAN + ]) + Style.BRIGHT + char + for char in msg + ) + else: # Default style + formatted_msg = msg + + print(formatted_msg) + +def encode_image(image_path): + """ + Encodes the image at the given path to a base64 string. + """ + with open(image_path, "rb") as image_file: + return base64.b64encode(image_file.read()).decode("utf-8") + +def capture_snapshot(debug=DEBUG_DEFAULT): + """ + Downloads a single snapshot JPEG from the cameraโ€™s snapshot URL + and returns it as a CV2 image (numpy array). + """ + # Use your cameraโ€™s IP and credentials from .env + snapshot_url = f"https://{CAMERA_IP}/cgi-bin/api.cgi?cmd=Snap&channel=0&rs=wuuPhkmUCeI9WG7C&user={CAMERA_USER}&password={CAMERA_PASS}" + + if debug: + print(f"Fetching snapshot from camera") #{snapshot_url}") + + # Disable SSL certificate verification for now; + # can add a proper certificate or turn verification on if desired. + response = requests.get(snapshot_url, verify=False) + response.raise_for_status() # Raise an error if request failed + + # Convert JPEG bytes to a numpy array + img_array = np.frombuffer(response.content, np.uint8) + # Decode the image using OpenCV + img = cv2.imdecode(img_array, cv2.IMREAD_COLOR) + + if img is None: + raise ValueError("Failed to decode the image from the camera.") + + if debug: + print("Snapshot captured successfully.") + + return img + # """Captures a single snapshot from the Reolink camera.""" + # # Using the reolinkapi + # cam = Camera(CAMERA_IP, CAMERA_USER, CAMERA_PASS, https=True) + # # This gets a stream generator; we'll just grab a single frame. + # stream = cam.open_video_stream() + # img = next(stream) + # debug_print("Captured image from camera.", debug) + # return img + +def crop_image(img, coords, debug=DEBUG_DEFAULT): + """Crops the image using the given coordinates. + Coordinates are (left, top, right, bottom).""" + left, top, right, bottom = coords + cropped_img = img[top:bottom, left:right] + debug_print(f"Cropped image with coords: {coords}", debug, "muted") + return cropped_img + +def save_image(img, timestamp, debug=DEBUG_DEFAULT): + """Saves the image with a filename based on the timestamp.""" + filename = os.path.join(IMAGES_FOLDER, f"{timestamp}.jpg") + cv2.imwrite(filename, img) + debug_print(f"Saved image to {filename}", debug, "muted") + return filename + +def log_event(message): + """Logs the given message with a timestamp.""" + logging.info(message) + +def submit_to_model(image_path, debug=DEBUG_DEFAULT): + """ + Submits the image to the OpenAI model to detect squirrels. + Returns True if a squirrel is detected, False otherwise. + """ + debug_print(f"Submitting {image_path} to model.", debug) + + # Initialize OpenAI client + client = OpenAI() + + try: + # Encode the image to base64 + base64_image = encode_image(image_path) + if debug: + debug_print(f"Image successfully encoded to base64.", debug) + + # Create the API request + response = client.chat.completions.create( + model="gpt-4o-mini-2024-07-18", + messages=[ + { + "role": "user", + "content": [ + { + "type": "text", + "text": "Is there a squirrel in the image? Answer with one word: yes or no.", + }, + { + "type": "image_url", + "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}, + }, + ], + } + ], + max_tokens=1000, + ) + + # Extract the response + answer = response.choices[0].message.content.strip().lower() + answer = answer.rstrip(".") + debug_print(f"Model response: {repr(answer)}", debug, "highlight") + + + return answer == "yes" + + except Exception as e: + debug_print(f"Error querying OpenAI API: {str(e)}", debug) + return False + + +def confirm_detection(debug=DEBUG_DEFAULT): + """After a positive detection, take 2 more snapshots quickly and confirm if at least one more is also positive.""" + debug_print("Confirming detection with 2 additional shots.", debug) + + for i in range(2): + img = capture_snapshot(debug) + # Crop the image + cropped = crop_image(img, CROP_COORDS, debug) + # Save the image + ts = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_f{i}" + image_path = save_image(cropped, ts, debug) + # Submit to model + if submit_to_model(image_path, debug): + debug_print(f"Snapshot {i+1}: Model response - yes", debug, "warn") + # Move file to the "positive_detection" sub-folder + new_path = os.path.join(POSITIVE_DETECTION_FOLDER, os.path.basename(image_path)) + shutil.move(image_path, new_path) + debug_print(f"Image moved to {new_path}", debug) + return True + else: + debug_print(f"Snapshot {i+1}: Model response - no", debug, "warn") + + # Small delay between confirmation shots if needed + # time.sleep(0.5) + + return False + +def play_alert(debug=DEBUG_DEFAULT): + """ + Plays an alert sound located in the DATA_FOLDER. + """ + alert_file = os.path.join(DATA_FOLDER, "scream.wav") + + if not os.path.exists(alert_file): + raise FileNotFoundError(f"Alert sound file not found: {alert_file}") + + if debug: + print(f"Playing alert sound from {alert_file}") + + # Initialize the mixer + pygame.mixer.init() + try: + # Load and play the sound + pygame.mixer.music.load(alert_file) + pygame.mixer.music.play() + + # Wait until the sound finishes + while pygame.mixer.music.get_busy(): + time.sleep(0.1) + except Exception as e: + print(f"Error playing alert sound: {e}") + finally: + pygame.mixer.quit() + +# Function to publish MQTT alert +def send_mqtt_alert(message, debug=DEBUG_DEFAULT): + """ + Publishes an alert message to the configured MQTT broker and topic. + """ + if not MQTT_BROKER or not MQTT_TOPIC: + raise ValueError("MQTT_BROKER or MQTT_TOPIC is not set. Check your .env file.") + + if debug: + print(f"Connecting to MQTT Broker at {MQTT_BROKER}:{MQTT_PORT}") + print(f"MQTT_USER: {MQTT_USERNAME}, MQTT_PASS: {MQTT_PASSWORD}") + print(f"MQTT_BROKER: {MQTT_BROKER}, MQTT_PORT: {MQTT_PORT} (type: {type(MQTT_PORT)})") + + client = mqtt.Client() + + if MQTT_USERNAME and MQTT_PASSWORD: + client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) + else: + print("MQTT username or password is missing. Check your .env file.") + + try: + client.connect(MQTT_BROKER, MQTT_PORT, 60) + if debug: + print(f"Publishing message to topic {MQTT_TOPIC}: {message}") + + client.publish(MQTT_TOPIC, message) + client.disconnect() + + if debug: + print("MQTT message sent successfully.") + except Exception as e: + if debug: + print(f"Failed to send MQTT message: {e}") + +def is_within_daylight(): + city = LocationInfo("Hilton Head Island", "US", "America/New_York", 32.155705183279615, -80.76296652972201) + s = sun(city.observer, date=datetime.now()) + + # Get the local timezone + local_tz = timezone(city.timezone) + + # Convert current time to offset-aware in the same timezone as `sun` results + now = datetime.now(local_tz) + + return s['sunrise'] <= now <= s['sunset'] + +def suns_out_buns_out(): + """ + Prints a colorful 'Suns Out, Buns Out' message with emojis using Colorama. + """ + sun_emoji = "โ˜€๏ธ" + peach_emoji = "๐Ÿ‘" + + # The obnoxious message with colors + message = ( + Fore.YELLOW + Style.BRIGHT + sun_emoji + + Fore.MAGENTA + Style.BRIGHT + " Suns Out, " + + Fore.YELLOW + Style.BRIGHT + sun_emoji + + Fore.CYAN + Style.BRIGHT + " Buns Out! " + + Fore.MAGENTA + peach_emoji + + Fore.YELLOW + sun_emoji + ) + print(message) + +def sleepy_desk_art(debug=DEBUG_DEFAULT, ai=False): + """ + Makes an OpenAI call to generate a fun sleepy phrase, then displays it with emojis. + Debugging is added to trace potential issues with the API call. + """ + if ai: + client = OpenAI() + debug_print("Starting OpenAI call to generate a sleepy phrase...", debug, "muted") + try: + # Make the OpenAI API call to get a sleepy phrase + response = client.chat.completions.create( + model="gpt-4", + messages=[ + { + "role": "user", + "content": "Write a fun and sleepy phrase in less than 50 characters." + } + ], + max_tokens=20, + ) + # Extract the phrase from the API response + phrase = response.choices[0].message.content.strip() + debug_print(f"OpenAI API response: {repr(phrase)}", debug, "highlight") + + except Exception as e: + # Log error details for debugging + error_message = f"OpenAI call failed: {str(e)}" + debug_print(error_message, debug, "danger") + # Use a default phrase in case of failure + phrase = "Dreaming of squirrels... ๐Ÿ’ค" + else: + phrase = "nut in my pussy daddy... ๐Ÿ’ค" + + # Display the phrase with emojis + art = ( + Fore.MAGENTA + Style.BRIGHT + + f""" + ๐Ÿ›Œ๐Ÿ’ค {phrase} ๐Ÿ’ค๐Ÿ˜ด + ๐Ÿ˜ด๐ŸŒ™ + """ + ) + debug_print("Displaying sleepy art message.", debug, "highlight") + print(art) + + + +# ========================== +# Main Loop +# ========================== + +def main(debug=True): + global iteration + iteration += 1 + print("Current iteration: " + str(iteration) + ".") + + debug_print("Starting single capture test...", debug) + debug_print("user is " + CAMERA_USER, debug) + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + + # Capture one image + img = capture_snapshot(debug) + + # Crop the image + cropped_img = crop_image(img, CROP_COORDS, debug) + + # Save the image + image_path = save_image(cropped_img, timestamp, debug) + + # Send to OpenAI for Processing + squirrel_detected = submit_to_model(image_path, debug) + debug_print(f"Squirrel detected status = {squirrel_detected} ", debug) + + debug_print(f"Test complete. Image saved at {image_path}", debug) + + if squirrel_detected: + debug_print("Squirrel detected. Initiating confirmation steps.", debug) + # Log preliminary detection + log_event(f"{timestamp}: Preliminary squirrel detection.") + + # Move file to the "positive_detection" sub-folder + new_path = os.path.join(POSITIVE_DETECTION_FOLDER, os.path.basename(image_path)) + shutil.move(image_path, new_path) + debug_print(f"Image moved to {new_path}", debug) + + # Confirm detection + if confirm_detection(debug): + log_event(f"{timestamp}: Confirmed squirrel detection.") + play_alert() + debug_print("Squirrel detection confirmed.", debug) + + # Send MQTT alert + alert_message = f"fire" + send_mqtt_alert(alert_message, debug) + + else: + log_event(f"{timestamp}: Detection not confirmed.") + debug_print("Squirrel detection not confirmed after additional checks.", debug) + else: + log_event(f"{timestamp}: No squirrel detected.") + + +if __name__ == "__main__": + + while True: + if is_within_daylight(): + suns_out_buns_out() + main() + else: + sleepy_desk_art() + + # Wait before next capture + time.sleep(CAPTURE_INTERVAL) + +