more cleanup for photos provider
This commit is contained in:
parent
c37bc6e60e
commit
4ed70eee90
3 changed files with 45 additions and 33 deletions
19
my/common.py
19
my/common.py
|
@ -126,3 +126,22 @@ def mcachew(*args, **kwargs):
|
||||||
import cachew.experimental
|
import cachew.experimental
|
||||||
cachew.experimental.enable_exceptions() # TODO do it only once?
|
cachew.experimental.enable_exceptions() # TODO do it only once?
|
||||||
return cachew.cachew(*args, **kwargs)
|
return cachew.cachew(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@functools.lru_cache(1)
|
||||||
|
def _magic():
|
||||||
|
import magic # type: ignore
|
||||||
|
return magic.Magic(mime=True)
|
||||||
|
|
||||||
|
|
||||||
|
# TODO could reuse in pdf module?
|
||||||
|
import mimetypes # TODO do I need init()?
|
||||||
|
def fastermime(path: str) -> str:
|
||||||
|
# mimetypes is faster
|
||||||
|
(mime, _) = mimetypes.guess_type(path)
|
||||||
|
if mime is not None:
|
||||||
|
return mime
|
||||||
|
# magic is slower but returns more stuff
|
||||||
|
# TODO FIXME Result type; it's inherently racey
|
||||||
|
return _magic().from_file(path)
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
"""
|
"""
|
||||||
Module for accessing photos and videos on the filesystem
|
Module for accessing photos and videos, with their GPS and timestamps
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# pip install geopy magic
|
# pip install geopy magic
|
||||||
|
@ -10,15 +10,14 @@ from pathlib import Path
|
||||||
from typing import Tuple, Dict, Optional, NamedTuple, Iterator, Iterable, List
|
from typing import Tuple, Dict, Optional, NamedTuple, Iterator, Iterable, List
|
||||||
|
|
||||||
from geopy.geocoders import Nominatim # type: ignore
|
from geopy.geocoders import Nominatim # type: ignore
|
||||||
import magic # type: ignore
|
|
||||||
|
|
||||||
from ..common import LazyLogger, mcachew
|
from ..common import LazyLogger, mcachew, fastermime
|
||||||
|
from ..error import Res
|
||||||
|
|
||||||
from mycfg import photos as config
|
from mycfg import photos as config
|
||||||
|
|
||||||
|
|
||||||
logger = LazyLogger('my.photos')
|
log = LazyLogger('my.photos')
|
||||||
log = logger
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@ -27,14 +26,11 @@ class LatLon(NamedTuple):
|
||||||
lat: float
|
lat: float
|
||||||
lon: float
|
lon: float
|
||||||
|
|
||||||
# TODO PIL.ExifTags.TAGS
|
|
||||||
|
|
||||||
|
|
||||||
class Photo(NamedTuple):
|
class Photo(NamedTuple):
|
||||||
path: str
|
path: str
|
||||||
dt: Optional[datetime]
|
dt: Optional[datetime]
|
||||||
geo: Optional[LatLon]
|
geo: Optional[LatLon]
|
||||||
# TODO can we always extract date? I guess not...
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def tags(self) -> List[str]: # TODO
|
def tags(self) -> List[str]: # TODO
|
||||||
|
@ -42,6 +38,7 @@ class Photo(NamedTuple):
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def _basename(self) -> str:
|
def _basename(self) -> str:
|
||||||
|
# TODO 'canonical' or something? only makes sense for organized ones
|
||||||
for bp in config.paths:
|
for bp in config.paths:
|
||||||
if self.path.startswith(bp):
|
if self.path.startswith(bp):
|
||||||
return self.path[len(bp):]
|
return self.path[len(bp):]
|
||||||
|
@ -54,12 +51,15 @@ class Photo(NamedTuple):
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def url(self) -> str:
|
def url(self) -> str:
|
||||||
|
PHOTOS_URL = 'TODO FIXME'
|
||||||
return PHOTOS_URL + self._basename
|
return PHOTOS_URL + self._basename
|
||||||
|
|
||||||
|
|
||||||
from .utils import get_exif_from_file, ExifTags, Exif, dt_from_path, convert_ref
|
from .utils import get_exif_from_file, ExifTags, Exif, dt_from_path, convert_ref
|
||||||
|
|
||||||
def _try_photo(photo: Path, mtype: str, *, parent_geo: Optional[LatLon]) -> Photo:
|
Result = Res[Photo]
|
||||||
|
|
||||||
|
def _make_photo(photo: Path, mtype: str, *, parent_geo: Optional[LatLon]) -> Iterator[Result]:
|
||||||
exif: Exif
|
exif: Exif
|
||||||
if any(x in mtype for x in {'image/png', 'image/x-ms-bmp', 'video'}):
|
if any(x in mtype for x in {'image/png', 'image/x-ms-bmp', 'video'}):
|
||||||
# TODO don't remember why..
|
# TODO don't remember why..
|
||||||
|
@ -94,7 +94,6 @@ def _try_photo(photo: Path, mtype: str, *, parent_geo: Optional[LatLon]) -> Phot
|
||||||
log.warning('ignoring timestamp extraction for %s, they are stupid for Instagram videos', photo)
|
log.warning('ignoring timestamp extraction for %s, they are stupid for Instagram videos', photo)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# TODO FIXME result type here??
|
|
||||||
edt = dt_from_path(photo) # ok, last try..
|
edt = dt_from_path(photo) # ok, last try..
|
||||||
|
|
||||||
if edt is None:
|
if edt is None:
|
||||||
|
@ -102,7 +101,7 @@ def _try_photo(photo: Path, mtype: str, *, parent_geo: Optional[LatLon]) -> Phot
|
||||||
|
|
||||||
if edt is not None and edt > datetime.now():
|
if edt is not None and edt > datetime.now():
|
||||||
# TODO also yield?
|
# TODO also yield?
|
||||||
logger.error('datetime for %s is too far in future: %s', photo, edt)
|
log.error('datetime for %s is too far in future: %s', photo, edt)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return edt
|
return edt
|
||||||
|
@ -110,24 +109,14 @@ def _try_photo(photo: Path, mtype: str, *, parent_geo: Optional[LatLon]) -> Phot
|
||||||
geo = _get_geo()
|
geo = _get_geo()
|
||||||
dt = _get_dt()
|
dt = _get_dt()
|
||||||
|
|
||||||
return Photo(str(photo), dt=dt, geo=geo)
|
yield Photo(str(photo), dt=dt, geo=geo)
|
||||||
|
|
||||||
|
|
||||||
import mimetypes # TODO do I need init()?
|
|
||||||
def fastermime(path: str, mgc=magic.Magic(mime=True)) -> str:
|
|
||||||
# mimetypes is faster
|
|
||||||
(mime, _) = mimetypes.guess_type(path)
|
|
||||||
if mime is not None:
|
|
||||||
return mime
|
|
||||||
# magic is slower but returns more stuff
|
|
||||||
# TODO FIXME Result type; it's inherently racey
|
|
||||||
return mgc.from_file(path)
|
|
||||||
|
|
||||||
|
|
||||||
# TODO exclude
|
# TODO exclude
|
||||||
def _candidates() -> Iterable[str]:
|
def _candidates() -> Iterable[str]:
|
||||||
# TODO that could be a bit slow if there are to many extra files?
|
# TODO that could be a bit slow if there are to many extra files?
|
||||||
from subprocess import Popen, PIPE
|
from subprocess import Popen, PIPE
|
||||||
|
# TODO could extract this to common?
|
||||||
with Popen([
|
with Popen([
|
||||||
'fdfind',
|
'fdfind',
|
||||||
'--follow',
|
'--follow',
|
||||||
|
@ -143,12 +132,12 @@ def _candidates() -> Iterable[str]:
|
||||||
continue
|
continue
|
||||||
if tp not in {'image', 'video'}:
|
if tp not in {'image', 'video'}:
|
||||||
# TODO yield error?
|
# TODO yield error?
|
||||||
logger.warning('%s: unexpected mime %s', path, tp)
|
log.warning('%s: unexpected mime %s', path, tp)
|
||||||
# TODO return mime too? so we don't have to call it again in _photos?
|
# TODO return mime too? so we don't have to call it again in _photos?
|
||||||
yield path
|
yield path
|
||||||
|
|
||||||
|
|
||||||
def photos() -> Iterator[Photo]:
|
def photos() -> Iterator[Result]:
|
||||||
candidates = tuple(sorted(_candidates()))
|
candidates = tuple(sorted(_candidates()))
|
||||||
return _photos(candidates)
|
return _photos(candidates)
|
||||||
# TODO figure out how to use cachew without helper function?
|
# TODO figure out how to use cachew without helper function?
|
||||||
|
@ -157,8 +146,8 @@ def photos() -> Iterator[Photo]:
|
||||||
|
|
||||||
# if geo information is missing from photo, you can specify it manually in geo.json file
|
# if geo information is missing from photo, you can specify it manually in geo.json file
|
||||||
# TODO is there something more standard?
|
# TODO is there something more standard?
|
||||||
# @mcachew(cache_path=config.cache_path)
|
@mcachew(cache_path=config.cache_path)
|
||||||
def _photos(candidates: Iterable[str]) -> Iterator[Photo]:
|
def _photos(candidates: Iterable[str]) -> Iterator[Result]:
|
||||||
geolocator = Nominatim() # TODO does it cache??
|
geolocator = Nominatim() # TODO does it cache??
|
||||||
|
|
||||||
from functools import lru_cache
|
from functools import lru_cache
|
||||||
|
@ -189,12 +178,14 @@ def _photos(candidates: Iterable[str]) -> Iterator[Photo]:
|
||||||
|
|
||||||
parent_geo = get_geo(path.parent)
|
parent_geo = get_geo(path.parent)
|
||||||
mime = fastermime(str(path))
|
mime = fastermime(str(path))
|
||||||
p = _try_photo(path, mime, parent_geo=parent_geo)
|
yield from _make_photo(path, mime, parent_geo=parent_geo)
|
||||||
yield p
|
|
||||||
|
|
||||||
|
|
||||||
def print_all():
|
def print_all():
|
||||||
for p in photos():
|
for p in photos():
|
||||||
|
if isinstance(p, Exception):
|
||||||
|
print('ERROR!', p)
|
||||||
|
else:
|
||||||
print(f"{p.dt} {p.path} {p.tags}")
|
print(f"{p.dt} {p.path} {p.tags}")
|
||||||
|
|
||||||
# TODO cachew -- improve AttributeError: type object 'tuple' has no attribute '__annotations__' -- improve errors?
|
# TODO cachew -- improve AttributeError: type object 'tuple' has no attribute '__annotations__' -- improve errors?
|
||||||
|
|
|
@ -1,3 +1,4 @@
|
||||||
|
from pathlib import Path
|
||||||
from typing import Dict
|
from typing import Dict
|
||||||
|
|
||||||
import PIL.Image # type: ignore
|
import PIL.Image # type: ignore
|
||||||
|
@ -6,6 +7,8 @@ from PIL.ExifTags import TAGS, GPSTAGS # type: ignore
|
||||||
|
|
||||||
Exif = Dict
|
Exif = Dict
|
||||||
|
|
||||||
|
# TODO PIL.ExifTags.TAGS
|
||||||
|
|
||||||
|
|
||||||
class ExifTags:
|
class ExifTags:
|
||||||
DATETIME = "DateTimeOriginal"
|
DATETIME = "DateTimeOriginal"
|
||||||
|
@ -17,9 +20,9 @@ class ExifTags:
|
||||||
|
|
||||||
|
|
||||||
# TODO there must be something more standard for this...
|
# TODO there must be something more standard for this...
|
||||||
def get_exif_from_file(path: str) -> Exif:
|
def get_exif_from_file(path: Path) -> Exif:
|
||||||
# TODO exception handler?
|
# TODO exception handler?
|
||||||
with PIL.Image.open(path) as fo:
|
with PIL.Image.open(str(path)) as fo:
|
||||||
return get_exif_data(fo)
|
return get_exif_data(fo)
|
||||||
|
|
||||||
|
|
||||||
|
@ -70,7 +73,6 @@ def convert_ref(cstr, ref: str):
|
||||||
|
|
||||||
import re
|
import re
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
# TODO surely there is a library that does it??
|
# TODO surely there is a library that does it??
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue