Coverage for cc_modules/cc_blob.py : 54%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
3"""
4camcops_server/cc_modules/cc_blob.py
6===============================================================================
8 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com).
10 This file is part of CamCOPS.
12 CamCOPS is free software: you can redistribute it and/or modify
13 it under the terms of the GNU General Public License as published by
14 the Free Software Foundation, either version 3 of the License, or
15 (at your option) any later version.
17 CamCOPS is distributed in the hope that it will be useful,
18 but WITHOUT ANY WARRANTY; without even the implied warranty of
19 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20 GNU General Public License for more details.
22 You should have received a copy of the GNU General Public License
23 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
25===============================================================================
27**BLOB (binary large object) handling.**
29"""
31import logging
32from typing import Optional, Type, TYPE_CHECKING
34from cardinal_pythonlib.httpconst import MimeType
35from cardinal_pythonlib.logs import BraceStyleAdapter
36from pendulum import DateTime as Pendulum
37from sqlalchemy.orm import relationship
38from sqlalchemy.orm import Session as SqlASession
39from sqlalchemy.orm.relationships import RelationshipProperty
40from sqlalchemy.sql.schema import Column
41from sqlalchemy.sql.sqltypes import Integer, Text
42import wand.image
44from camcops_server.cc_modules.cc_db import (
45 GenericTabletRecordMixin,
46 TaskDescendant,
47)
48from camcops_server.cc_modules.cc_html import (
49 get_data_url,
50 get_embedded_img_tag,
51)
52from camcops_server.cc_modules.cc_simpleobjects import TaskExportOptions
53from camcops_server.cc_modules.cc_sqla_coltypes import (
54 CamcopsColumn,
55 MimeTypeColType,
56 TableNameColType,
57)
58from camcops_server.cc_modules.cc_sqla_coltypes import (
59 LongBlob,
60 RelationshipInfo,
61)
62from camcops_server.cc_modules.cc_sqlalchemy import Base
63from camcops_server.cc_modules.cc_xml import (
64 get_xml_blob_element,
65 XmlElement,
66)
68if TYPE_CHECKING:
69 from camcops_server.cc_modules.cc_request import CamcopsRequest # noqa: E501,F401
70 from camcops_server.cc_modules.cc_task import Task # noqa: F401
72log = BraceStyleAdapter(logging.getLogger(__name__))
74# ExactImage API documentation is a little hard to find. See:
75# http://www.exactcode.com/site/open_source/exactimage
76# man econvert # after sudo apt-get install exactimage
77# https://exactcode.de/exact-image/trunk/api/api.hh <-- THIS ONE
78# http://fossies.org/linux/privat/exact-image-0.8.9.tar.gz:a/exact-image-0.8.9/examples/test.py # noqa
79# http://lickmychip.com/2012/07/26/playing-with-exactimage/
80# https://github.com/romainneutron/ExactImage-PHP
81# Also, rotation is not simple!
82# Wand seems much better: http://docs.wand-py.org/en/0.3.5/
85# =============================================================================
86# Blob class
87# =============================================================================
89class Blob(GenericTabletRecordMixin, TaskDescendant, Base):
90 """
91 Class representing a binary large object (BLOB).
93 Has helper functions for PNG image processing.
94 """
95 __tablename__ = "blobs"
96 id = Column(
97 "id", Integer,
98 nullable=False,
99 comment="BLOB (binary large object) primary key on the source "
100 "tablet device"
101 )
102 tablename = Column(
103 "tablename", TableNameColType,
104 nullable=False,
105 comment="Name of the table referring to this BLOB"
106 )
107 tablepk = Column(
108 "tablepk", Integer,
109 nullable=False,
110 comment="Client-perspective primary key (id field) of the row "
111 "referring to this BLOB"
112 )
113 fieldname = Column(
114 "fieldname", TableNameColType,
115 nullable=False,
116 comment="Field name of the field referring to this BLOB by ID"
117 )
118 filename = CamcopsColumn(
119 "filename", Text, # Text is correct; filenames can be long
120 exempt_from_anonymisation=True,
121 comment="Filename of the BLOB on the source tablet device (on "
122 "the source device, BLOBs are stored in files, not in "
123 "the database)"
124 )
125 mimetype = Column(
126 "mimetype", MimeTypeColType,
127 comment="MIME type of the BLOB"
128 )
129 image_rotation_deg_cw = Column(
130 "image_rotation_deg_cw", Integer,
131 comment="For images: rotation to be applied, clockwise, in degrees"
132 )
133 theblob = Column(
134 "theblob", LongBlob,
135 comment="The BLOB itself, a binary object containing arbitrary "
136 "information (such as a picture)"
137 ) # type: Optional[bytes]
139 @classmethod
140 def get_current_blob_by_client_info(cls,
141 dbsession: SqlASession,
142 device_id: int,
143 clientpk: int,
144 era: str) -> Optional['Blob']:
145 """
146 Returns the current Blob object, or None.
147 """
148 # noinspection PyPep8
149 blob = (
150 dbsession.query(cls)
151 .filter(cls.id == clientpk)
152 .filter(cls._device_id == device_id)
153 .filter(cls._era == era)
154 .filter(cls._current == True) # noqa: E712
155 .first()
156 ) # type: Optional[Blob]
157 return blob
159 @classmethod
160 def get_contemporaneous_blob_by_client_info(
161 cls,
162 dbsession: SqlASession,
163 device_id: int,
164 clientpk: int,
165 era: str,
166 referrer_added_utc: Pendulum,
167 referrer_removed_utc: Optional[Pendulum]) \
168 -> Optional['Blob']:
169 """
170 Returns a contemporaneous Blob object, or None.
172 Use particularly to look up BLOBs matching old task records.
173 """
174 blob = (
175 dbsession.query(cls)
176 .filter(cls.id == clientpk)
177 .filter(cls._device_id == device_id)
178 .filter(cls._era == era)
179 .filter(cls._when_added_batch_utc <= referrer_added_utc)
180 .filter(cls._when_removed_batch_utc == referrer_removed_utc)
181 .first()
182 ) # type: Optional[Blob]
183 # Note, for referrer_removed_utc: if this is None, then the comparison
184 # "field == None" is made; otherwise "field == value".
185 # Since SQLAlchemy translates "== None" to "IS NULL", we're OK.
186 # https://stackoverflow.com/questions/37445041/sqlalchemy-how-to-filter-column-which-contains-both-null-and-integer-values # noqa
187 return blob
189 def get_rotated_image(self) -> Optional[bytes]:
190 """
191 Returns a binary image, having rotated if necessary, or None.
192 """
193 if not self.theblob:
194 return None
195 rotation = self.image_rotation_deg_cw
196 if rotation is None or rotation % 360 == 0:
197 return self.theblob
198 with wand.image.Image(blob=self.theblob) as img:
199 img.rotate(rotation)
200 return img.make_blob()
201 # ... no parameter => return in same format as supplied
203 def get_img_html(self) -> str:
204 """
205 Returns an HTML IMG tag encoding the BLOB, or ''.
206 """
207 image_bits = self.get_rotated_image()
208 if not image_bits:
209 return ""
210 return get_embedded_img_tag(self.mimetype or MimeType.PNG, image_bits)
211 # Historically, CamCOPS supported only PNG, so add this as a default
213 def get_xml_element(self, req: "CamcopsRequest") -> XmlElement:
214 """
215 Returns an :class:`camcops_server.cc_modules.cc_xml.XmlElement`
216 representing this BLOB.
217 """
218 options = TaskExportOptions(xml_skip_fields=["theblob"],
219 xml_include_plain_columns=True,
220 include_blobs=False)
221 branches = self._get_xml_branches(req, options)
222 blobdata = self._get_xml_theblob_value_binary()
223 branches.append(get_xml_blob_element(
224 name="theblob",
225 blobdata=blobdata,
226 comment=Blob.theblob.comment
227 ))
228 return XmlElement(name=self.__tablename__, value=branches)
230 def _get_xml_theblob_value_binary(self) -> Optional[bytes]:
231 """
232 Returns a binary value for this object, to be encoded into XML.
233 """
234 image_bits = self.get_rotated_image()
235 return image_bits
237 def get_data_url(self) -> str:
238 """
239 Returns a data URL encapsulating the BLOB, or ''.
240 """
241 if not self.theblob:
242 return ""
243 return get_data_url(self.mimetype or MimeType.PNG, self.theblob)
245 # -------------------------------------------------------------------------
246 # TaskDescendant overrides
247 # -------------------------------------------------------------------------
249 @classmethod
250 def task_ancestor_class(cls) -> Optional[Type["Task"]]:
251 return None
253 def task_ancestor(self) -> Optional["Task"]:
254 from camcops_server.cc_modules.cc_task import tablename_to_task_class_dict # noqa # delayed import
255 d = tablename_to_task_class_dict()
256 try:
257 cls = d[self.tablename] # may raise KeyError
258 return cls.get_linked(self.tablepk, self)
259 except KeyError:
260 return None
263# =============================================================================
264# Relationships
265# =============================================================================
267def blob_relationship(classname: str,
268 blob_id_col_attr_name: str,
269 read_only: bool = True) -> RelationshipProperty:
270 """
271 Simplifies creation of BLOB relationships.
272 In a class definition, use like this:
274 .. code-block:: python
276 class Something(Base):
278 photo_blobid = CamcopsColumn(
279 "photo_blobid", Integer,
280 is_blob_id_field=True, blob_field_xml_name="photo_blob"
281 )
283 photo = blob_relationship("Something", "photo_blobid")
285 # ... can't use Something directly as it's not yet been fully
286 # defined, but we want the convenience of defining this
287 # relationship here without the need to use metaclasses.
288 # ... SQLAlchemy's primaryjoin uses Python-side names (class and
289 # attribute), rather than SQL-side names (table and column),
290 # at least for its fancier things:
291 # https://docs.sqlalchemy.org/en/latest/orm/join_conditions.html#relationship-primaryjoin # noqa
293 Note that this refers to the CURRENT version of the BLOB. If there is
294 an editing chain, older BLOB versions are not retrieved.
296 Compare :class:`camcops_server.cc_modules.cc_task.TaskHasPatientMixin`,
297 which uses the same strategy, as do several other similar functions.
299 """
300 return relationship(
301 Blob,
302 primaryjoin=(
303 "and_("
304 " remote(Blob.id) == foreign({cls}.{fk}), "
305 " remote(Blob._device_id) == foreign({cls}._device_id), "
306 " remote(Blob._era) == foreign({cls}._era), "
307 " remote(Blob._current) == True "
308 ")".format(
309 cls=classname,
310 fk=blob_id_col_attr_name
311 )
312 ),
313 uselist=False,
314 viewonly=read_only,
315 info={
316 RelationshipInfo.IS_BLOB: True,
317 },
318 )
321# =============================================================================
322# Unit tests
323# =============================================================================
325def get_blob_img_html(blob: Optional[Blob],
326 html_if_missing: str = "<i>(No picture)</i>") -> str:
327 """
328 For the specified BLOB, get an HTML IMG tag with embedded data, or an HTML
329 error message.
330 """
331 if blob is None:
332 return html_if_missing
333 return blob.get_img_html() or html_if_missing