Coverage for cc_modules/cc_blob.py: 53%
83 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-15 14:23 +0100
1"""
2camcops_server/cc_modules/cc_blob.py
4===============================================================================
6 Copyright (C) 2012, University of Cambridge, Department of Psychiatry.
7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).
9 This file is part of CamCOPS.
11 CamCOPS is free software: you can redistribute it and/or modify
12 it under the terms of the GNU General Public License as published by
13 the Free Software Foundation, either version 3 of the License, or
14 (at your option) any later version.
16 CamCOPS is distributed in the hope that it will be useful,
17 but WITHOUT ANY WARRANTY; without even the implied warranty of
18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 GNU General Public License for more details.
21 You should have received a copy of the GNU General Public License
22 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
24===============================================================================
26**BLOB (binary large object) handling.**
28"""
30import logging
31from typing import Optional, Type, TYPE_CHECKING
33from cardinal_pythonlib.httpconst import MimeType
34from cardinal_pythonlib.logs import BraceStyleAdapter
35from pendulum import DateTime as Pendulum
36from sqlalchemy.orm import relationship
37from sqlalchemy.orm import Mapped, mapped_column, Session as SqlASession
38from sqlalchemy.orm.relationships import RelationshipProperty
39from sqlalchemy.sql.sqltypes import Integer, Text
40import wand.image
42from camcops_server.cc_modules.cc_db import (
43 GenericTabletRecordMixin,
44 TaskDescendant,
45)
46from camcops_server.cc_modules.cc_html import (
47 get_data_url,
48 get_embedded_img_tag,
49)
50from camcops_server.cc_modules.cc_simpleobjects import TaskExportOptions
51from camcops_server.cc_modules.cc_sqla_coltypes import (
52 mapped_camcops_column,
53 MimeTypeColType,
54 TableNameColType,
55)
56from camcops_server.cc_modules.cc_sqla_coltypes import (
57 LongBlob,
58 RelationshipInfo,
59)
60from camcops_server.cc_modules.cc_sqlalchemy import Base
61from camcops_server.cc_modules.cc_xml import get_xml_blob_element, XmlElement
63if TYPE_CHECKING:
64 from camcops_server.cc_modules.cc_request import (
65 CamcopsRequest,
66 )
67 from camcops_server.cc_modules.cc_task import Task
69log = BraceStyleAdapter(logging.getLogger(__name__))
71# ExactImage API documentation is a little hard to find. See:
72# http://www.exactcode.com/site/open_source/exactimage
73# man econvert # after sudo apt-get install exactimage
74# https://exactcode.de/exact-image/trunk/api/api.hh <-- THIS ONE
75# http://fossies.org/linux/privat/exact-image-0.8.9.tar.gz:a/exact-image-0.8.9/examples/test.py # noqa
76# http://lickmychip.com/2012/07/26/playing-with-exactimage/
77# https://github.com/romainneutron/ExactImage-PHP
78# Also, rotation is not simple!
79# Wand seems much better: http://docs.wand-py.org/en/0.3.5/
82# =============================================================================
83# Blob class
84# =============================================================================
87class Blob(GenericTabletRecordMixin, TaskDescendant, Base):
88 """
89 Class representing a binary large object (BLOB).
91 Has helper functions for PNG image processing.
92 """
94 __tablename__ = "blobs"
95 id: Mapped[int] = mapped_column(
96 comment="BLOB (binary large object) primary key on the source "
97 "tablet device",
98 )
99 tablename: Mapped[str] = mapped_column(
100 TableNameColType,
101 comment="Name of the table referring to this BLOB",
102 )
103 tablepk: Mapped[int] = mapped_column(
104 comment="Client-perspective primary key (id field) of the row "
105 "referring to this BLOB",
106 )
107 fieldname: Mapped[str] = mapped_column(
108 TableNameColType,
109 comment="Field name of the field referring to this BLOB by ID",
110 )
111 filename: Mapped[Optional[str]] = mapped_camcops_column(
112 Text, # Text is correct; filenames can be long
113 exempt_from_anonymisation=True,
114 comment="Filename of the BLOB on the source tablet device (on "
115 "the source device, BLOBs are stored in files, not in "
116 "the database)",
117 )
118 mimetype: Mapped[Optional[str]] = mapped_column(
119 MimeTypeColType, comment="MIME type of the BLOB"
120 )
121 image_rotation_deg_cw: Mapped[Optional[int]] = mapped_column(
122 "image_rotation_deg_cw",
123 Integer,
124 comment="For images: rotation to be applied, clockwise, in degrees",
125 )
126 theblob: Mapped[Optional[bytes]] = mapped_column(
127 "theblob",
128 LongBlob,
129 comment="The BLOB itself, a binary object containing arbitrary "
130 "information (such as a picture)",
131 )
133 @classmethod
134 def get_current_blob_by_client_info(
135 cls, dbsession: SqlASession, device_id: int, clientpk: int, era: str
136 ) -> Optional["Blob"]:
137 """
138 Returns the current Blob object, or None.
139 """
140 # noinspection PyPep8
141 blob = (
142 dbsession.query(cls)
143 .filter(cls.id == clientpk) # type: ignore[arg-type]
144 .filter(cls._device_id == device_id)
145 .filter(cls._era == era)
146 .filter(cls._current == True) # noqa: E712
147 .first()
148 ) # type: Optional[Blob]
149 return blob
151 @classmethod
152 def get_contemporaneous_blob_by_client_info(
153 cls,
154 dbsession: SqlASession,
155 device_id: int,
156 clientpk: int,
157 era: str,
158 referrer_added_utc: Pendulum,
159 referrer_removed_utc: Optional[Pendulum],
160 ) -> Optional["Blob"]:
161 """
162 Returns a contemporaneous Blob object, or None.
164 Use particularly to look up BLOBs matching old task records.
165 """
166 blob = (
167 dbsession.query(cls)
168 .filter(cls.id == clientpk) # type: ignore[arg-type]
169 .filter(cls._device_id == device_id)
170 .filter(cls._era == era)
171 .filter(cls._when_added_batch_utc <= referrer_added_utc)
172 .filter(cls._when_removed_batch_utc == referrer_removed_utc)
173 .first()
174 ) # type: Optional[Blob]
175 # Note, for referrer_removed_utc: if this is None, then the comparison
176 # "field == None" is made; otherwise "field == value".
177 # Since SQLAlchemy translates "== None" to "IS NULL", we're OK.
178 # https://stackoverflow.com/questions/37445041/sqlalchemy-how-to-filter-column-which-contains-both-null-and-integer-values # noqa
179 return blob
181 def get_rotated_image(self) -> Optional[bytes]:
182 """
183 Returns a binary image, having rotated if necessary, or None.
184 """
185 if not self.theblob:
186 return None
187 rotation = self.image_rotation_deg_cw
188 if rotation is None or rotation % 360 == 0:
189 return self.theblob
190 with wand.image.Image(blob=self.theblob) as img:
191 img.rotate(rotation)
192 return img.make_blob()
193 # ... no parameter => return in same format as supplied
195 def get_img_html(self) -> str:
196 """
197 Returns an HTML IMG tag encoding the BLOB, or ''.
198 """
199 image_bits = self.get_rotated_image()
200 if not image_bits:
201 return ""
202 return get_embedded_img_tag(self.mimetype or MimeType.PNG, image_bits)
203 # Historically, CamCOPS supported only PNG, so add this as a default
205 def get_xml_element(self, req: "CamcopsRequest") -> XmlElement:
206 """
207 Returns an :class:`camcops_server.cc_modules.cc_xml.XmlElement`
208 representing this BLOB.
209 """
210 options = TaskExportOptions(
211 xml_skip_fields=["theblob"],
212 xml_include_plain_columns=True,
213 include_blobs=False,
214 )
215 branches = self._get_xml_branches(req, options)
216 blobdata = self._get_xml_theblob_value_binary()
217 branches.append(
218 get_xml_blob_element(
219 name="theblob",
220 blobdata=blobdata,
221 comment=Blob.theblob.comment, # type: ignore[attr-defined]
222 )
223 )
224 return XmlElement(name=self.__tablename__, value=branches)
226 def _get_xml_theblob_value_binary(self) -> Optional[bytes]:
227 """
228 Returns a binary value for this object, to be encoded into XML.
229 """
230 image_bits = self.get_rotated_image()
231 return image_bits
233 def get_data_url(self) -> str:
234 """
235 Returns a data URL encapsulating the BLOB, or ''.
236 """
237 if not self.theblob:
238 return ""
239 return get_data_url(self.mimetype or MimeType.PNG, self.theblob)
241 # -------------------------------------------------------------------------
242 # TaskDescendant overrides
243 # -------------------------------------------------------------------------
245 @classmethod
246 def task_ancestor_class(cls) -> Optional[Type["Task"]]:
247 return None
249 def task_ancestor(self) -> Optional["Task"]:
250 from camcops_server.cc_modules.cc_task import (
251 tablename_to_task_class_dict,
252 ) # delayed import
254 d = tablename_to_task_class_dict()
255 try:
256 cls = d[self.tablename] # may raise KeyError
257 return cls.get_linked(self.tablepk, self)
258 except KeyError:
259 return None
262# =============================================================================
263# Relationships
264# =============================================================================
267def blob_relationship(
268 classname: str, blob_id_col_attr_name: str, read_only: bool = True
269) -> 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 = camcops_column(
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(cls=classname, fk=blob_id_col_attr_name)
309 ),
310 uselist=False,
311 viewonly=read_only,
312 info={RelationshipInfo.IS_BLOB: True},
313 )
316# =============================================================================
317# Unit tests
318# =============================================================================
321def get_blob_img_html(
322 blob: Optional[Blob], html_if_missing: str = "<i>(No picture)</i>"
323) -> str:
324 """
325 For the specified BLOB, get an HTML IMG tag with embedded data, or an HTML
326 error message.
327 """
328 if blob is None:
329 return html_if_missing
330 return blob.get_img_html() or html_if_missing