import os, time, ssl, binascii, wifi, vectorio, socketpool, adafruit_requests, displayio from jpegio import JpegDecoder from adafruit_display_text import label, wrap_text_to_lines import terminalio import adafruit_pycamera import adafruit_minimqtt.adafruit_minimqtt as MQTT # scale for displaying returned text from OpenAI text_scale = 2 trash = ("Can you identify what the object is and then classify what type of trash it is " "(Organic, Glass, Paper, Metal, Plastic, Hazard, Compost, Landfill)? " "Follow this template: The object appears to be ____. This type of trash for this item is ___.") prompts = [trash] num_prompts = len(prompts) prompt_index = 0 prompt_labels = ["CLASSIFY TRASH"] # encode jpeg to base64 for OpenAI def encode_image(image_path): with open(image_path, 'rb') as image_file: image_data = image_file.read() base64_encoded_data = binascii.b2a_base64(image_data).decode('utf-8').rstrip() return base64_encoded_data # view returned text on MEMENTO screen def view_text(the_text): rectangle = vectorio.Rectangle(pixel_shader=palette, width=240, height=240, x=0, y=0) pycam.splash.append(rectangle) the_text = "\n".join(wrap_text_to_lines(the_text, 20)) if prompt_index == 1: the_text = the_text.replace("*", "\n") text_area = label.Label(terminalio.FONT, text=the_text, color=0xFFFFFF, x=2, y=10, scale=text_scale) pycam.splash.append(text_area) pycam.display.refresh() # send image to OpenAI, print the returned text and save it as a text file def send_img(img, prompt): base64_image = encode_image(img) headers = { "Content-Type": "application/json", "Authorization": f"Bearer API-KEY" } payload = { "model": "gpt-4-turbo", "messages": [ { "role": "user", "content": [ { "type": "text", "text": f"{prompts[prompt_index]}" }, { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{base64_image}" } } ] } ], "max_tokens": 300 } response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload) json_openai = response.json() print(response.json()) TEXT = json_openai['choices'][0]['message']['content'] if "Organic" in TEXT: mqtt_client.publish(light_sound_feed, "organic") if "Metal" in TEXT: mqtt_client.publish(light_sound_feed, "metal") if "Glass" in TEXT: mqtt_client.publish(light_sound_feed, "glass") if "Paper" in TEXT: mqtt_client.publish(light_sound_feed, "paper") if "Plastic" in TEXT: mqtt_client.publish(light_sound_feed, "plastic") if "Hazard" in TEXT: mqtt_client.publish(light_sound_feed, "hazard") if "Landfill" in TEXT: mqtt_client.publish(light_sound_feed, "landfill") if "Compost" in TEXT: mqtt_client.publish(light_sound_feed, "compost") alt_text_file = img.replace('jpg', 'txt') alt_text_file = alt_text_file[:11] + f"_{prompt_labels[prompt_index]}" + alt_text_file[11:] if prompt_index == 5: alt_text_file = alt_text_file.replace("?", "") with open(alt_text_file, "a") as fp: fp.write(json_openai['choices'][0]['message']['content']) fp.flush() time.sleep(1) fp.close() view_text(json_openai['choices'][0]['message']['content']) # view images on sd card to re-send to OpenAI def load_image(bit, file): bit.fill(0b00000_000000_00000) # fill with a middle grey decoder.open(file) decoder.decode(bit, scale=0, x=0, y=0) pycam.blit(bit, y_offset=32) pycam.display.refresh() print() print("Connecting to WiFi") try: wifi.radio.connect('WIFI_SSID', 'WIFI_PASSWORD') print(f"Connected to Wi-Fi! IP address: {wifi.radio.ipv4_address}") except ConnectionError as e: print(f"Connection failed: {e}") print("Connected to WiFi") pool = socketpool.SocketPool(wifi.radio) requests = adafruit_requests.Session(pool, ssl.create_default_context()) mqtt_client = MQTT.MQTT( broker="io.adafruit.com", port=1883, username="parkbpx", password="aio_oXHQ14VNN4srK7hgTQI3dGbSoBgI", socket_pool=pool, ssl_context=ssl.create_default_context(), ) def connected(client, userdata, flags, rc): # Connected to broker at adafruit io print("Connected to Adafruit IO! Listening for topic changes in feeds I've subscribed to") # Subscribe to all changes on the feed. client.subscribe(light_sound_feed) def disconnected(client, userdata, rc): # Disconnected from the broker at adafruit io print("Disconnected from Adafruit IO!") def message(client, topic, message): print(f"Received message on topic {topic}: {message}") light_sound_feed = "parkbpx/feeds/light_sound_feed" mqtt_client.on_connect = connected mqtt_client.on_disconnect = disconnected mqtt_client.on_message = message mqtt_client.connect() palette = displayio.Palette(1) palette[0] = 0x000000 decoder = JpegDecoder() # used for showing images from sd card bitmap = displayio.Bitmap(240, 176, 65535) pycam = adafruit_pycamera.PyCamera() pycam.mode = 0 # only mode 0 (JPEG) will work in this example # Resolution of 320x240 is plenty for OpenAI pycam.resolution = 1 # 0-12 preset resolutions: # 0: 240x240, 1: 320x240, 2: 640x480, 3: 800x600, 4: 1024x768, # 5: 1280x720, 6: 1280x1024, 7: 1600x1200, 8: 1920x1080, 9: 2048x1536, # 10: 2560x1440, 11: 2560x1600, 12: 2560x1920 # pycam.led_level = 1 # 0-4 preset brightness levels # pycam.led_color = 0 # 0-7 preset colors: 0: white, 1: green, 2: yellow, 3: red, # 4: pink, 5: blue, 6: teal, 7: rainbow pycam.effect = 0 # 0-7 preset FX: 0: normal, 1: invert, 2: b&w, 3: red, # 4: green, 5: blue, 6: sepia, 7: solarize # sort image files by numeric order all_images = [ f"/sd/{filename}" for filename in os.listdir("/sd") if filename.lower().endswith(".jpg") ] all_images.sort(key=lambda f: int(''.join(filter(str.isdigit, f)))) # add label for selected prompt rect = vectorio.Rectangle(pixel_shader=palette, width=240, height=20, x=0, y=0) prompt_txt = label.Label( terminalio.FONT, text=prompt_labels[prompt_index], color=0xFF0055, x=10, y=15, scale=2 ) # pylint: disable=protected-access pycam._botbar.append(rect) pycam._botbar.append(prompt_txt) # pylint: enable=protected-access pycam.display.refresh() view = False new_prompt = False file_index = -1 while True: if new_prompt: pycam.display_message("SEND?") if not view: if not new_prompt: pycam.blit(pycam.continuous_capture()) pycam.keys_debounce() if pycam.shutter.long_press: pycam.autofocus() if pycam.shutter.short_count: try: pycam.display_message("snap", color=0x00DD00) pycam.capture_jpeg() pycam.live_preview_mode() except TypeError as exception: pycam.display_message("Failed", color=0xFF0000) time.sleep(0.5) pycam.live_preview_mode() except RuntimeError as exception: pycam.display_message("Error\nNo SD Card", color=0xFF0000) time.sleep(0.5) all_images = [ f"/sd/{filename}" for filename in os.listdir("/sd") if filename.lower().endswith(".jpg") ] all_images.sort(key=lambda f: int(''.join(filter(str.isdigit, f)))) the_image = all_images[-1] pycam.display_message("OpenAI..", color=0x00DD00) send_img(the_image, prompts[prompt_index]) view = True if pycam.up.fell: prompt_index = (prompt_index - 1) % num_prompts prompt_txt.text = prompt_labels[prompt_index] pycam.display.refresh() if pycam.down.fell: prompt_index = (prompt_index + 1) % num_prompts prompt_txt.text = prompt_labels[prompt_index] pycam.display.refresh() if pycam.right.fell: if new_prompt: file_index = (file_index - -1) % -len(all_images) filename = all_images[file_index] load_image(bitmap, filename) else: prompt_index = (prompt_index + 1) % num_prompts prompt_txt.text = prompt_labels[prompt_index] pycam.display.refresh() if pycam.left.fell: if new_prompt: file_index = (file_index + -1) % -len(all_images) filename = all_images[file_index] load_image(bitmap, filename) else: prompt_index = (prompt_index - 1) % num_prompts prompt_txt.text = prompt_labels[prompt_index] pycam.display.refresh() if pycam.select.fell: if not new_prompt: file_index = -1 new_prompt = True filename = all_images[file_index] load_image(bitmap, filename) else: new_prompt = False pycam.display.refresh() if pycam.ok.fell: if view: pycam.splash.pop() pycam.splash.pop() pycam.display.refresh() view = False if new_prompt: pycam.display_message("OpenAI..", color=0x00DD00) send_img(filename, prompts[prompt_index]) new_prompt = False view = True # mqtt_client.loop(2)