Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
# -*- coding: utf-8 -*-
import tempfile
from contextlib import contextmanager
from requests import HTTPError
from arkindex_worker import logger
from arkindex_worker.image import download_tiles, open_image, polygon_bounding_box
class MagicDict(dict):
"""
A dict whose items can be accessed like attributes.
"""
def _magify(self, item):
"""
Automagically convert lists and dicts to MagicDicts and lists of MagicDicts
Allows for nested access: foo.bar.baz
"""
if isinstance(item, list):
return list(map(self._magify, item))
if isinstance(item, dict):
return MagicDict(item)
return item
def __getitem__(self, item):
item = super().__getitem__(item)
return self._magify(item)
def __getattr__(self, name):
try:
return self[name]
except KeyError:
raise AttributeError(
"{} object has no attribute '{}'".format(self.__class__.__name__, name)
)
def __delattr__(self, name):
try:
return super().__delattr__(name)
except AttributeError:
try:
return super().__delitem__(name)
except KeyError:
pass
raise
def __dir__(self):
return super().__dir__() + list(self.keys())
class Element(MagicDict):
"""
Describes any kind of element.
"""
def image_url(self, size="full"):
"""
When possible, will return the S3 URL for images, so an ML worker can bypass IIIF servers
:param size: Subresolution of the image.
"""
if not self.get("zone"):
return
url = self.zone.image.get("s3_url")
if url:
return url
url = self.zone.image.url
if not url.endswith("/"):
url += "/"
return "{}full/{}/0/default.jpg".format(url, size)
@property
def requires_tiles(self):
if not self.get("zone") or self.zone.image.get("s3_url"):
return False
max_width = self.zone.image.server.max_width or float("inf")
max_height = self.zone.image.server.max_height or float("inf")
bounding_box = polygon_bounding_box(self.zone.polygon)
return bounding_box.width > max_width or bounding_box.height > max_height
def open_image(self, *args, max_size=None, **kwargs):
"""
Open this element's image as a Pillow image.
This does not crop the image to the element's polygon.
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
:param max_size: Subresolution of the image.
"""
if not self.get("zone"):
raise ValueError("Element {} has no zone".format(self.id))
if self.requires_tiles:
if max_size is None:
return download_tiles(self.zone.image.url)
else:
raise NotImplementedError
if max_size is not None:
bounding_box = polygon_bounding_box(self.zone.polygon)
original_size = {"w": self.zone.image.width, "h": self.zone.image.height}
# No resizing if the element is smaller than the image.
if (
bounding_box.width != original_size["w"]
or bounding_box.height != original_size["h"]
):
resize = "full"
logger.warning(
"Only full image size elements covered, "
+ "downloading full size image."
)
# No resizing if the image is smaller than the wanted size.
elif original_size["w"] <= max_size and original_size["h"] <= max_size:
resize = "full"
# Resizing if the image is bigger than the wanted size.
else:
ratio = max_size / max(original_size.values())
new_width, new_height = [int(x * ratio) for x in original_size.values()]
resize = "{},{}".format(new_width, new_height)
else:
resize = "full"
try:
return open_image(self.image_url(resize), *args, **kwargs)
except HTTPError as e:
if (
self.zone.image.get("s3_url") is not None
and e.response.status_code == 403
):
# This element uses an S3 URL: the URL may have expired.
# Call the API to get a fresh URL and try again
# TODO: this should be done by the worker
raise NotImplementedError
return open_image(self.image_url(resize), *args, **kwargs)
raise
@contextmanager
def open_image_tempfile(self, format="jpeg", *args, **kwargs):
"""
Get the element's image as a temporary file stored on the disk.
To be used as a context manager: with element.open_image_tempfile() as f: ...
"""
with tempfile.NamedTemporaryFile() as f:
self.open_image(*args, **kwargs).save(f, format=format)
yield f
def __str__(self):
if isinstance(self.type, dict):
type_name = self.type["display_name"]
else:
type_name = str(self.type)
return "{} {} ({})".format(type_name, self.name, self.id)