#Final Project import adafruit_requests import board, time, neopixel import os, ssl, socketpool, wifi import adafruit_minimqtt.adafruit_minimqtt as MQTT import rtc, circuitpython_schedule as schedule import gc #NEOPIXEL SETUP: strip = neopixel.NeoPixel(board.GP15, 30) from adafruit_led_animation.color import ( AMBER, #(255, 100, 0) AQUA, # (50, 255, 255) BLACK, #OFF (0, 0, 0) BLUE, # (0, 0, 255) CYAN, # (0, 255, 255) GOLD, # (255, 222, 30) GREEN, # (0, 255, 0) JADE, # (0, 255, 40) MAGENTA, #(255, 0, 20) OLD_LACE, # (253, 245, 230) ORANGE, # (255, 40, 0) PINK, # (242, 90, 255) PURPLE, # (180, 0, 255) RED, # (255, 0, 0) TEAL, # (0, 255, 120) WHITE, # (255, 255, 255) YELLOW, # (255, 150, 0) RAINBOW # a list of colors to cycle through # RAINBOW is RED, ORANGE, YELLOW, GREEN, BLUE, and PURPLE ((255, 0, 0), (255, 40, 0), (255, 150, 0), (0, 255, 0), (0, 0, 255), (180, 0, 255)) ) strip_color = PURPLE strip.fill(BLACK) #Set up Requests: pool = socketpool.SocketPool(wifi.radio) requests = adafruit_requests.Session(pool, ssl.create_default_context()) print(f"Connecting to WiFi") wifi.radio.connect(os.getenv("WIFI_SSID"), os.getenv("WIFI_PASSWORD")) print("Connected") #TIME SETUP: clock = rtc.RTC() #gc.collect() def get_time(): print(f"Accessing url: https://worldtimeapi.org/api/timezone/America/Chicago") print(gc.mem_free()) #gc.collect() response = requests.get('https://worldtimeapi.org/api/timezone/America/Chicago') gc.collect() # convert response into json json = response.json() # parse out values that we want unixtime = json["unixtime"] #raw_offset = json["raw_offset"] (does not take off-cycle DST into account so I did not use this line) raw_offset = -14400 #Hardcoded version - this gets the accurate EST while DST is not in effect, change to -18000 for DST # create the local time in seconds location_time = unixtime + raw_offset # turn seconds into time components current_time = time.localtime(location_time) # format and print time & date printable_time = f"{current_time.tm_hour:d}:{current_time.tm_min:02d}:{current_time.tm_sec:02d}" printable_date = f"{current_time.tm_mon:d}/{current_time.tm_mday:d}/{current_time.tm_year:02d}" print(f"printable_time: {printable_time}") print(f"printable_date: {printable_date}") # set the rtc with the component time in current_time clock.datetime = time.struct_time(current_time) #get_time() #schedule.every().day.at("00:00").do(get_time) #WEATHER API SETUP: site = 'http://api.openweathermap.org/data/2.5/weather' key = 'Your Key Here' location = "Your Zip Code Here" units = "units=imperial" url2 = (f"{site}?{location}&{key}&{units}") print(url2) #Fucntion to Call to Weather API and set LEDs according to weather forecast def get_weather(): #gc.collect() print("getting weather") print(gc.mem_free()) count = 0 response2 = requests.get(url2) json = response2.json() forecast = (json['weather'][0]['main']) low = (json['main']['temp_min']) high = (json['main']['temp_max']) temp = (float(low) + float(high))/2 print(forecast) print(temp) strip.brightness = 1.0 if forecast == 'Rain' or forecast == 'Thunderstorm': #For heavier rains - wear a rain jacket! strip.fill(BLUE) time.sleep(10) strip.fill(BLACK) elif forecast == "Clouds": #For lighter rain - Pack a light rain jacket strip.fill(AQUA) time.sleep(10) strip.fill(BLACK) elif forecast == "Mist" or forecast == "mist": #For mist/fog - Wear a windbreaker strip.fill(CYAN) time.sleep(10) strip.fill(BLACK) elif forecast == "Snow": #Self explanatory - grab a hat and boots! strip.fill(OLD_LACE) time.sleep(10) strip.fill(BLACK) elif temp <= 35: #It's gonna be cold today - dress warm strip.fill(RED) time.sleep(10) strip.fill(BLACK) elif temp <= 60: #It's gonna be a little chilly, wear pants and a sweatshirt to be safe strip.fill(YELLOW) time.sleep(10) strip.fill(BLACK) elif temp > 60: #it's over 60! strip.fill(GREEN) time.sleep(10) strip.fill(BLACK) else: if count < 20: #flash some colors to remind me to check the weather - the forecast must be weird today! strip.fill(random.choice(RAINBOW)) time.sleep(10) count = count + 1 else: strip.fill(BLACK) #CONNECTING TO DASHBOARD: aio_username = os.getenv("AIO_USERNAME") aio_key = os.getenv("AIO_KEY") cloud_on_off = aio_username + "/feeds/cloud_on_off" cloud_color = aio_username + "/feeds/cloud_color" time_feed = aio_username + "/feeds/time_feed" def connected(client, userdata, flags, rc): print("Connected to Adafruit IO! Listening for topic changes in feed.") client.subscribe(cloud_on_off) client.subscribe(cloud_color) client.subscribe(time_feed) def disconnected(client, userdata, rc): print("Disconnected from Adafruit IO") def message(client, topic, message): global strip_color print(f"topic: {topic}, message: {message}") if topic == cloud_on_off: if message == "ON": print("Color should change now") strip.fill(strip_color) strip.brightness = 1.0 elif message == "OFF": print("Color should turn off now") strip.brightness = 0.0 elif topic == cloud_color: if message[0] == "#": message = message[1:] strip_color = int(message, 16) strip.fill(strip_color) elif topic == time_feed: try: alarm_time = message #print(alarm_time) #schedule.every().day.at(alarm_time).do(get_weather) #print(f"scheduled weather at: {alarm_time}") get_weather() except: pass mqtt_client = MQTT.MQTT( broker = os.getenv("BROKER"), port = os.getenv("PORT"), username = aio_username, password = aio_key, socket_pool = pool, ssl_context = ssl.create_default_context() ) mqtt_client.on_connect = connected mqtt_client.on_disconnect = disconnected mqtt_client.on_message = message print("Connecting to Adafruit IO...") mqtt_client.connect() while True: mqtt_client.loop() schedule.run_pending()