Compare commits

...

2 Commits

2 changed files with 75 additions and 33 deletions

View File

@ -1,3 +1,12 @@
# EXIF Filter
**0.2.0:**
- Drop requirement on GPSPhoto. Use only exifread.
- Extract time of image creation from EXIF.
**0.1.0:**
- Initial release.
# OpenStreetMap Tool
**2.2.1:**
- Round distances to 3 decimal places.

99
exif.py
View File

@ -2,23 +2,29 @@
title: EXIF Filter
author: projectmoon
author_url: https://git.agnos.is/projectmoon/open-webui-filters
version: 0.1.0
version: 0.2.0
license: AGPL-3.0+
required_open_webui_version: 0.6.5
requirements: pillow, piexif, exifread, gpsphoto
requirements: exifread
"""
import requests
from urllib.parse import urljoin
import os
from GPSPhoto import gpsphoto
from pydantic import BaseModel, Field
from typing import Callable, Awaitable, Any, Optional, Literal
import json
from base64 import b64decode
import os
import tempfile
from base64 import b64decode
from io import BytesIO
from typing import Any, Awaitable, Callable, Literal, Optional
from urllib.parse import urljoin
import requests
from exifread import process_file
from open_webui.utils.misc import (
add_or_update_system_message,
get_last_user_message,
get_last_user_message_item,
)
from pydantic import BaseModel, Field
from open_webui.utils.misc import get_last_user_message_item, get_last_user_message, add_or_update_system_message
def get_or_none(tags: dict, *keys: str) -> Optional[str]:
"""
@ -62,7 +68,6 @@ def parse_nominatim_address(address) -> Optional[str]:
line3 = ", ".join(line3).strip()
full_address = filter(None, [line1, line2, line3])
full_address = ", ".join(full_address).strip()
print(full_address)
return full_address if len(full_address) > 0 else None
class OsmCache:
@ -189,23 +194,38 @@ class OsmSearcher:
return place_name
def extract_gps(img_bytes):
with tempfile.NamedTemporaryFile(delete=False) as fp:
fp.write(img_bytes)
fp.close()
def convert_to_decimal(tags, gps_tag, gps_ref_tag):
if gps_tag not in tags or gps_ref_tag not in tags:
return None
try:
data = gpsphoto.getGPSData(fp.name)
lat = data.get('Latitude', None)
lon = data.get('Longitude', None)
os.unlink(fp.name)
values = tags[gps_tag].values
ref = tags[gps_ref_tag].values[0]
if lat and lon:
return (round(lat, 4), round(lon, 4))
else:
return None
except Exception as e:
print(f"[EXIF-OSM] WARNING: Could not load image for GPS processing: {e}")
degrees = sum(
values[i].numerator / values[i].denominator * (1/(60**i))
for i in range(3)
)
return -degrees if (ref == 'W' and gps_tag == 'GPSLongitude') or \
(ref == 'S' and gps_tag == 'GPSLatitude') else degrees
def extract_exif_info(img_bytes):
try:
f = BytesIO(img_bytes)
tags = process_file(f, strict=False)
date_taken = tags.get('EXIF DateTimeOriginal', None)
lat = convert_to_decimal(tags, 'GPS GPSLatitude', 'GPS GPSLatitudeRef')
lon = convert_to_decimal(tags, 'GPS GPSLongitude', 'GPS GPSLongitudeRef')
if lon is not None and lat is not None:
coords = (round(lat, 4), round(lon, 4))
else:
coords = None
return { "date_taken": date_taken, "gps_coords": coords }
except Exception as e:
print(f"[EXIF-OSM] WARNING: Could not load image for GPS processing: {e}")
return None
def exif_instructions(geocoding, user_image_count):
if geocoding:
@ -217,11 +237,17 @@ def valid_instructions(geocoding, user_image_count):
lat = geocoding.get("lat", "unknown")
lon = geocoding.get("lon", "unknown")
place_name = geocoding.get("place", None)
date_taken = geocoding.get("date_taken", None)
if date_taken:
date_inst = f"This photo was taken on {date_taken}."
else:
date_inst = ""
if place_name:
place_inst = f"The location (accurate to radius of 5 to 10 meters) is: {place_name}"
place_inst = f"The location (accurate to radius of 5 to 10 meters) is: {place_name}."
else:
place_inst = "The name of the location could not be determined"
place_inst = "The name of the location could not be determined."
count_inst = (f"There are {user_image_count} images from the user in this chat. "
f"The most recent image is image number {user_image_count}.")
@ -234,7 +260,7 @@ def valid_instructions(geocoding, user_image_count):
return (f"\n\nYou have access to GPS location information about the "
f"most recent image in this chat. The image's GPS coordinates "
f"are: {lat},{lon}. {place_inst}. {osm_inst}"
f"are: {lat},{lon}. {place_inst} {date_inst} {osm_inst}"
f"\n\nThis applies to ONLY the most recent image in the chat. {count_inst}")
def invalid_instructions():
@ -292,13 +318,20 @@ class Filter:
self.valves = self.Valves()
async def process_image(self, image_data_url):
base64_img = image_data_url.split(',')[1]
base64_img = image_data_url.split(',', maxsplit=1)[1]
img_bytes = b64decode(base64_img)
coords = extract_gps(img_bytes)
if coords:
exif = extract_exif_info(img_bytes)
if exif:
coords = exif["gps_coords"]
searcher = OsmSearcher(self.valves)
geocoded_name = await searcher.reverse_geocode(coords[0], coords[1])
return { "lat": coords[0], "lon": coords[1], "place": geocoded_name }
return {
"date_taken": exif["date_taken"],
"place": geocoded_name,
"lat": coords[0],
"lon": coords[1]
}
else:
return None