Compare commits
2 Commits
814a639dfe
...
0b24aeb0a1
Author | SHA1 | Date |
---|---|---|
|
0b24aeb0a1 | |
|
b0dea19eb9 |
|
@ -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
99
exif.py
|
@ -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
|
||||
|
||||
|
|
Loading…
Reference in New Issue