tykayn/mapillary_download
3
0
Fork 3
Code Issues 2 Pull Requests Actions Projects Releases Wiki Activity

Merge pull request #5 from Stefal/oom_fix

Browse Source 
Oom fix
This commit is contained in:
Stefal 2023-09-17 20:08:22 +02:00 committed by GitHub
commit 45ea0df8b6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 190 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
download.py
View File

@ -36,11 +36,13 @@ def background(f):
return wrapped return wrapped
#@background #@background
def download(url, filepath, metadata=None): def download(url, filepath, metadata=None):
#print(asizeof.asizeof(image)/1024, "MB")
with open(str(filepath), "wb") as f: with open(str(filepath), "wb") as f:
r = session.get(url, stream=True, timeout=6) r = session.get(url, stream=True, timeout=6)
image = write_exif(r.content, metadata) image = write_exif(r.content, metadata)
f.write(image) f.write(image)
#del image
print("{} downloaded".format(filepath)) print("{} downloaded".format(filepath))
def get_single_image_data(image_id, mly_header): def get_single_image_data(image_id, mly_header):
@ -97,11 +99,14 @@ def write_exif(picture, img_metadata):
#{'thumb_original_url': 'https://scontent-cdg4-2.xx.fbcdn.net/m1/v/t6/An9Zy2SrH9vXJIF01QkBODyUbg7XSKfwL48UwHyvihSwvECGjVbG0vSw9uhxe2-Dq-k2eUcigb83buO6zo-7eVbykfp5aQIe1kgd-MJr66nU_H-o_mwBLZXgVbj5I_5WX-C9c6FxJruHkV962F228O0?ccb=10-5&oh=00_AfDOKD869DxL-4ZNCbVo8Rn29vsc0JyjMAU2ctx4aAFVMQ&oe=65256C25&_nc_sid=201bca', #{'thumb_original_url': 'https://scontent-cdg4-2.xx.fbcdn.net/m1/v/t6/An9Zy2SrH9vXJIF01QkBODyUbg7XSKfwL48UwHyvihSwvECGjVbG0vSw9uhxe2-Dq-k2eUcigb83buO6zo-7eVbykfp5aQIe1kgd-MJr66nU_H-o_mwBLZXgVbj5I_5WX-C9c6FxJruHkV962F228O0?ccb=10-5&oh=00_AfDOKD869DxL-4ZNCbVo8Rn29vsc0JyjMAU2ctx4aAFVMQ&oe=65256C25&_nc_sid=201bca',
# 'captured_at': 1603459736644, 'geometry': {'type': 'Point', 'coordinates': [2.5174596904057, 48.777089857534]}, 'id': '485924785946693'} # 'captured_at': 1603459736644, 'geometry': {'type': 'Point', 'coordinates': [2.5174596904057, 48.777089857534]}, 'id': '485924785946693'}
picture = writer.writePictureMetadata(picture, img_metadata) with writer.Writer(picture) as image:
picture = writer.add_altitude(picture, img_metadata) image.writePictureMetadata(img_metadata)
picture = writer.add_direction(picture, img_metadata) image.add_altitude(img_metadata)
image.add_direction(img_metadata)
image.apply()
updated_image = image.get_Bytes()
return picture return updated_image
if __name__ == '__main__': if __name__ == '__main__':
parse_args() parse_args()

384
writer.py
View File

@ -28,228 +28,204 @@ class PictureMetadata:
direction: Optional[float] = None direction: Optional[float] = None
orientation: Optional[int] = 1 orientation: Optional[int] = 1
class Writer():
def __init__(self, picture: bytes):
self.content = picture
self.image = pyexiv2.ImageData(picture)
self.exif = self.image.read_exif()
self.xmp = self.image.read_xmp()
self.updated_exif = {}
self.updated_xmp = {}
def writePictureMetadata(picture: bytes, metadata: PictureMetadata) -> bytes: def __enter__(self):
""" return self
Override exif metadata on raw picture and return updated bytes
"""
if not metadata.capture_time and not metadata.longitude and not metadata.latitude and not metadata.picture_type:
return picture
if metadata.capture_time:
picture = add_gps_datetime(picture, metadata)
picture = add_datetimeoriginal(picture, metadata)
if metadata.latitude is not None and metadata.longitude is not None:
picture = add_lat_lon(picture, metadata)
if metadata.picture_type is not None:
picture = add_img_projection(picture, metadata)
return picture
def add_lat_lon(picture: bytes, metadata: PictureMetadata) -> bytes:
"""
Add latitude and longitude values in GPSLatitude + GPSLAtitudeRef and GPSLongitude + GPSLongitudeRef
"""
img = pyexiv2.ImageData(picture)
updated_exif = {}
if metadata.latitude is not None:
updated_exif["Exif.GPSInfo.GPSLatitudeRef"] = "N" if metadata.latitude > 0 else "S"
updated_exif["Exif.GPSInfo.GPSLatitude"] = _to_exif_dms(metadata.latitude)
if metadata.longitude is not None:
updated_exif["Exif.GPSInfo.GPSLongitudeRef"] = "E" if metadata.longitude > 0 else "W"
updated_exif["Exif.GPSInfo.GPSLongitude"] = _to_exif_dms(metadata.longitude)
if updated_exif: def __exit__(self, *args):
img.modify_exif(updated_exif) self.image.close()
def apply(self):
self.image.modify_exif(self.updated_exif)
self.image.modify_xmp(self.updated_xmp)
def close(self):
self.image.close()
def get_Bytes(self):
return self.image.get_bytes()
return img.get_bytes() def writePictureMetadata(self, metadata: PictureMetadata):
"""
Override exif metadata on raw picture and return updated bytes
"""
if not metadata.capture_time and not metadata.longitude and not metadata.latitude and not metadata.picture_type:
return
def add_altitude(picture: bytes, metadata: PictureMetadata, precision: int = 1000) -> bytes: if metadata.capture_time:
""" self.add_gps_datetime(metadata)
Add altitude value in GPSAltitude and GPSAltitudeRef self.add_datetimeoriginal(metadata)
"""
altitude = metadata.altitude
img = pyexiv2.ImageData(picture)
updated_exif = {}
if altitude is not None: if metadata.latitude is not None and metadata.longitude is not None:
negative_altitude = 0 if altitude >= 0 else 1 self.add_lat_lon(metadata)
updated_exif['Exif.GPSInfo.GPSAltitude'] = f"{int(abs(altitude * precision))} / {precision}"
updated_exif['Exif.GPSInfo.GPSAltitudeRef'] = negative_altitude
if updated_exif: if metadata.picture_type is not None:
img.modify_exif(updated_exif) self.add_img_projection(metadata)
return img.get_bytes() def add_lat_lon(self, metadata: PictureMetadata):
"""
Add latitude and longitude values in GPSLatitude + GPSLAtitudeRef and GPSLongitude + GPSLongitudeRef
"""
if metadata.latitude is not None:
self.updated_exif["Exif.GPSInfo.GPSLatitudeRef"] = "N" if metadata.latitude > 0 else "S"
self.updated_exif["Exif.GPSInfo.GPSLatitude"] = self._to_exif_dms(metadata.latitude)
if metadata.longitude is not None:
self.updated_exif["Exif.GPSInfo.GPSLongitudeRef"] = "E" if metadata.longitude > 0 else "W"
self.updated_exif["Exif.GPSInfo.GPSLongitude"] = self._to_exif_dms(metadata.longitude)
def add_altitude(self, metadata: PictureMetadata, precision: int = 1000):
"""
Add altitude value in GPSAltitude and GPSAltitudeRef
"""
altitude = metadata.altitude
if altitude is not None:
negative_altitude = 0 if altitude >= 0 else 1
self.updated_exif['Exif.GPSInfo.GPSAltitude'] = f"{int(abs(altitude * precision))} / {precision}"
self.updated_exif['Exif.GPSInfo.GPSAltitudeRef'] = negative_altitude
def add_direction(self, metadata: PictureMetadata, ref: str = 'T', precision: int = 1000):
"""
Add direction value in GPSImgDirection and GPSImgDirectionRef
"""
direction = metadata.direction
if metadata.direction is not None:
self.updated_exif['Exif.GPSInfo.GPSImgDirection'] = f"{int(abs(direction % 360.0 * precision))} / {precision}"
self.updated_exif['Exif.GPSInfo.GPSImgDirectionRef'] = ref
def add_gps_datetime(self, metadata: PictureMetadata):
"""
Add GPSDateStamp and GPSTimeStamp
"""
if metadata.capture_time.utcoffset() is None:
metadata.capture_time = self.localize(metadata)
# for capture time, override GPSInfo time and DatetimeOriginal
self.updated_exif["Exif.Photo.DateTimeOriginal"] = metadata.capture_time.strftime("%Y:%m:%d %H:%M:%S")
offset = metadata.capture_time.utcoffset()
if offset is not None:
self.updated_exif["Exif.Photo.OffsetTimeOriginal"] = self.format_offset(offset)
utc_dt = metadata.capture_time.astimezone(tz=pytz.UTC)
self.updated_exif["Exif.GPSInfo.GPSDateStamp"] = utc_dt.strftime("%Y:%m:%d")
self.updated_exif["Exif.GPSInfo.GPSTimeStamp"] = utc_dt.strftime("%H/1 %M/1 %S/1")
def add_datetimeoriginal(self, metadata: PictureMetadata):
"""
Add date time in Exif DateTimeOriginal and SubSecTimeOriginal tags
"""
if metadata.capture_time.utcoffset() is None:
metadata.capture_time = self.localize(metadata)
# for capture time, override DatetimeOriginal and SubSecTimeOriginal
self.updated_exif["Exif.Photo.DateTimeOriginal"] = metadata.capture_time.strftime("%Y:%m:%d %H:%M:%S")
offset = metadata.capture_time.utcoffset()
if offset is not None:
self.updated_exif["Exif.Photo.OffsetTimeOriginal"] = self.format_offset(offset)
if metadata.capture_time.microsecond != 0:
self.updated_exif["Exif.Photo.SubSecTimeOriginal"] = metadata.capture_time.strftime("%f")
def add_img_projection(self, metadata: PictureMetadata):
"""
Add image projection type (equirectangular for spherical image, ...) in xmp GPano.ProjectionType
"""
if metadata.picture_type.value != "flat":
self.updated_xmp["Xmp.GPano.ProjectionType"] = metadata.picture_type.value
self.updated_xmp["Xmp.GPano.UsePanoramaViewer"] = True
def format_offset(self, offset: timedelta) -> str:
"""Format offset for OffsetTimeOriginal. Format is like "+02:00" for paris offset
>>> format_offset(timedelta(hours=5, minutes=45))
'+05:45'
>>> format_offset(timedelta(hours=-3))
'-03:00'
"""
offset_hour, remainer = divmod(offset.total_seconds(), 3600)
return f"{'+' if offset_hour >= 0 else '-'}{int(abs(offset_hour)):02}:{int(remainer/60):02}"
def add_direction(picture: bytes, metadata: PictureMetadata, ref: str = 'T', precision: int = 1000) -> bytes: def localize(self, metadata: PictureMetadata) -> datetime:
""" """
Add direction value in GPSImgDirection and GPSImgDirectionRef Localize a datetime in the timezone of the picture
""" If the picture does not contains GPS position, the datetime will not be modified.
direction = metadata.direction """
img = pyexiv2.ImageData(picture) exif = self.exif
updated_exif = {} try:
lon = exif["Exif.GPSInfo.GPSLongitude"]
lon_ref = exif.get("Exif.GPSInfo.GPSLongitudeRef", "E")
lat = exif["Exif.GPSInfo.GPSLatitude"]
lat_ref = exif.get("Exif.GPSInfo.GPSLatitudeRef", "N")
except KeyError:
return metadata.capture_time # canot localize, returning same date
if metadata.direction is not None: lon = self._from_dms(lon) * (1 if lon_ref == "E" else -1)
updated_exif['Exif.GPSInfo.GPSImgDirection'] = f"{int(abs(direction % 360.0 * precision))} / {precision}" lat = self._from_dms(lat) * (1 if lat_ref == "N" else -1)
updated_exif['Exif.GPSInfo.GPSImgDirectionRef'] = ref
if updated_exif: tz_name = tz_finder.timezone_at(lng=lon, lat=lat)
img.modify_exif(updated_exif) if not tz_name:
return metadata.capture_time # cannot find timezone, returning same date
return img.get_bytes() tz = pytz.timezone(tz_name)
return tz.localize(metadata.capture_time)
def add_gps_datetime(picture: bytes, metadata: PictureMetadata) -> bytes: def _from_dms(self, val: str) -> float:
""" """Convert exif lat/lon represented as degre/minute/second into decimal
Add GPSDateStamp and GPSTimeStamp >>> _from_dms("1/1 55/1 123020/13567")
""" 1.9191854417991367
img = pyexiv2.ImageData(picture) >>> _from_dms("49/1 0/1 1885/76")
updated_exif = {} 49.00688961988304
"""
deg_raw, min_raw, sec_raw = val.split(" ")
deg_num, deg_dec = deg_raw.split("/")
deg = float(deg_num) / float(deg_dec)
min_num, min_dec = min_raw.split("/")
min = float(min_num) / float(min_dec)
sec_num, sec_dec = sec_raw.split("/")
sec = float(sec_num) / float(sec_dec)
if metadata.capture_time.utcoffset() is None: return float(deg) + float(min) / 60 + float(sec) / 3600
metadata.capture_time = localize(metadata, img)
# for capture time, override GPSInfo time and DatetimeOriginal
updated_exif["Exif.Photo.DateTimeOriginal"] = metadata.capture_time.strftime("%Y:%m:%d %H:%M:%S")
offset = metadata.capture_time.utcoffset()
if offset is not None:
updated_exif["Exif.Photo.OffsetTimeOriginal"] = format_offset(offset)
utc_dt = metadata.capture_time.astimezone(tz=pytz.UTC)
updated_exif["Exif.GPSInfo.GPSDateStamp"] = utc_dt.strftime("%Y:%m:%d")
updated_exif["Exif.GPSInfo.GPSTimeStamp"] = utc_dt.strftime("%H/1 %M/1 %S/1")
if updated_exif:
img.modify_exif(updated_exif)
return img.get_bytes()
def add_datetimeoriginal(picture: bytes, metadata: PictureMetadata) -> bytes:
"""
Add date time in Exif DateTimeOriginal and SubSecTimeOriginal tags
"""
img = pyexiv2.ImageData(picture)
updated_exif = {}
if metadata.capture_time.utcoffset() is None:
metadata.capture_time = localize(metadata, img)
# for capture time, override DatetimeOriginal and SubSecTimeOriginal
updated_exif["Exif.Photo.DateTimeOriginal"] = metadata.capture_time.strftime("%Y:%m:%d %H:%M:%S")
offset = metadata.capture_time.utcoffset()
if offset is not None:
updated_exif["Exif.Photo.OffsetTimeOriginal"] = format_offset(offset)
if metadata.capture_time.microsecond != 0:
updated_exif["Exif.Photo.SubSecTimeOriginal"] = metadata.capture_time.strftime("%f")
if updated_exif:
img.modify_exif(updated_exif)
return img.get_bytes()
def add_img_projection(picture: bytes, metadata: PictureMetadata) -> bytes:
"""
Add image projection type (equirectangular for spherical image, ...) in xmp GPano.ProjectionType
"""
img = pyexiv2.ImageData(picture)
updated_xmp = {}
if metadata.picture_type.value != "flat":
updated_xmp["Xmp.GPano.ProjectionType"] = metadata.picture_type.value
updated_xmp["Xmp.GPano.UsePanoramaViewer"] = True
if updated_xmp:
img.modify_xmp(updated_xmp)
return img.get_bytes()
def format_offset(offset: timedelta) -> str:
"""Format offset for OffsetTimeOriginal. Format is like "+02:00" for paris offset
>>> format_offset(timedelta(hours=5, minutes=45))
'+05:45'
>>> format_offset(timedelta(hours=-3))
'-03:00'
"""
offset_hour, remainer = divmod(offset.total_seconds(), 3600)
return f"{'+' if offset_hour >= 0 else '-'}{int(abs(offset_hour)):02}:{int(remainer/60):02}"
def localize(metadata: PictureMetadata, imagedata: pyexiv2.ImageData) -> datetime: def _to_dms(self, value: float) -> Tuple[int, int, float]:
""" """Return degree/minute/seconds for a decimal
Localize a datetime in the timezone of the picture >>> _to_dms(38.889469)
If the picture does not contains GPS position, the datetime will not be modified. (38, 53, 22.0884)
""" >>> _to_dms(43.7325)
exif = imagedata.read_exif() (43, 43, 57.0)
try: >>> _to_dms(-43.7325)
lon = exif["Exif.GPSInfo.GPSLongitude"] (43, 43, 57.0)
lon_ref = exif.get("Exif.GPSInfo.GPSLongitudeRef", "E") """
lat = exif["Exif.GPSInfo.GPSLatitude"] value = abs(value)
lat_ref = exif.get("Exif.GPSInfo.GPSLatitudeRef", "N") deg = int(value)
except KeyError: min = (value - deg) * 60
return metadata.capture_time # canot localize, returning same date sec = (min - int(min)) * 60
lon = _from_dms(lon) * (1 if lon_ref == "E" else -1) return deg, int(min), round(sec, 8)
lat = _from_dms(lat) * (1 if lat_ref == "N" else -1)
tz_name = tz_finder.timezone_at(lng=lon, lat=lat)
if not tz_name:
return metadata.capture_time # cannot find timezone, returning same date
tz = pytz.timezone(tz_name)
return tz.localize(metadata.capture_time)
def _from_dms(val: str) -> float: def _to_exif_dms(self, value: float) -> str:
"""Convert exif lat/lon represented as degre/minute/second into decimal """Return degree/minute/seconds string formated for the exif metadata for a decimal
>>> _from_dms("1/1 55/1 123020/13567") >>> _to_exif_dms(38.889469)
1.9191854417991367 '38/1 53/1 55221/2500'
>>> _from_dms("49/1 0/1 1885/76") """
49.00688961988304 from fractions import Fraction
"""
deg_raw, min_raw, sec_raw = val.split(" ")
deg_num, deg_dec = deg_raw.split("/")
deg = float(deg_num) / float(deg_dec)
min_num, min_dec = min_raw.split("/")
min = float(min_num) / float(min_dec)
sec_num, sec_dec = sec_raw.split("/")
sec = float(sec_num) / float(sec_dec)
return float(deg) + float(min) / 60 + float(sec) / 3600 d, m, s = self._to_dms(value)
f = Fraction.from_float(s).limit_denominator() # limit fraction precision
num_s, denomim_s = f.as_integer_ratio()
def _to_dms(value: float) -> Tuple[int, int, float]: return f"{d}/1 {m}/1 {num_s}/{denomim_s}"
"""Return degree/minute/seconds for a decimal
>>> _to_dms(38.889469)
(38, 53, 22.0884)
>>> _to_dms(43.7325)
(43, 43, 57.0)
>>> _to_dms(-43.7325)
(43, 43, 57.0)
"""
value = abs(value)
deg = int(value)
min = (value - deg) * 60
sec = (min - int(min)) * 60
return deg, int(min), round(sec, 8)
def _to_exif_dms(value: float) -> str:
"""Return degree/minute/seconds string formated for the exif metadata for a decimal
>>> _to_exif_dms(38.889469)
'38/1 53/1 55221/2500'
"""
from fractions import Fraction
d, m, s = _to_dms(value)
f = Fraction.from_float(s).limit_denominator() # limit fraction precision
num_s, denomim_s = f.as_integer_ratio()
return f"{d}/1 {m}/1 {num_s}/{denomim_s}"