Coverage for cc_modules/cc_patient.py: 36%
394 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-15 15:51 +0100
1"""
2camcops_server/cc_modules/cc_patient.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**Patients.**
28"""
30import datetime
31import logging
32from typing import (
33 Any,
34 Dict,
35 Generator,
36 List,
37 Optional,
38 Set,
39 Tuple,
40 TYPE_CHECKING,
41 Union,
42)
43import uuid as python_uuid
45from cardinal_pythonlib.classes import classproperty
46from cardinal_pythonlib.datetimefunc import (
47 coerce_to_pendulum_date,
48 format_datetime,
49 get_age,
50 PotentialDatetimeType,
51)
52from cardinal_pythonlib.json.typing_helpers import JsonObjectType
53from cardinal_pythonlib.logs import BraceStyleAdapter
54import cardinal_pythonlib.rnc_web as ws
55from fhirclient.models.address import Address
56from fhirclient.models.contactpoint import ContactPoint
57from fhirclient.models.humanname import HumanName
58from fhirclient.models.fhirreference import FHIRReference
59from fhirclient.models.identifier import Identifier
60from fhirclient.models.patient import Patient as FhirPatient
61import hl7
62import pendulum
63from sqlalchemy.ext.declarative import declared_attr
64from sqlalchemy.orm import mapped_column, Mapped, relationship
65from sqlalchemy.orm import Session as SqlASession
66from sqlalchemy.orm.relationships import RelationshipProperty
67from sqlalchemy.sql.expression import and_, ClauseElement, select
68from sqlalchemy.sql.selectable import SelectBase
69from sqlalchemy.sql.sqltypes import UnicodeText
71from camcops_server.cc_modules.cc_audit import audit
72from camcops_server.cc_modules.cc_constants import (
73 DateFormat,
74 ERA_NOW,
75 FHIRConst as Fc,
76 FP_ID_DESC,
77 FP_ID_SHORT_DESC,
78 FP_ID_NUM,
79 SEX_FEMALE,
80 SEX_MALE,
81 SEX_OTHER_UNSPECIFIED,
82 SPREADSHEET_PATIENT_FIELD_PREFIX,
83)
84from camcops_server.cc_modules.cc_dataclasses import SummarySchemaInfo
85from camcops_server.cc_modules.cc_db import GenericTabletRecordMixin
86from camcops_server.cc_modules.cc_fhir import (
87 fhir_pk_identifier,
88 make_fhir_bundle_entry,
89)
90from camcops_server.cc_modules.cc_hl7 import make_pid_segment
91from camcops_server.cc_modules.cc_html import answer
92from camcops_server.cc_modules.cc_idnumdef import IdNumDefinition
93from camcops_server.cc_modules.cc_simpleobjects import (
94 BarePatientInfo,
95 HL7PatientIdentifier,
96)
97from camcops_server.cc_modules.cc_patientidnum import (
98 extra_id_colname,
99 PatientIdNum,
100)
101from camcops_server.cc_modules.cc_proquint import proquint_from_uuid
102from camcops_server.cc_modules.cc_report import Report
103from camcops_server.cc_modules.cc_simpleobjects import (
104 IdNumReference,
105 TaskExportOptions,
106)
107from camcops_server.cc_modules.cc_specialnote import SpecialNote
108from camcops_server.cc_modules.cc_sqla_coltypes import (
109 EmailAddressColType,
110 mapped_camcops_column,
111 PatientNameColType,
112 SexColType,
113 UuidColType,
114)
115from camcops_server.cc_modules.cc_sqlalchemy import Base
116from camcops_server.cc_modules.cc_spreadsheet import SpreadsheetPage
117from camcops_server.cc_modules.cc_version import CAMCOPS_SERVER_VERSION_STRING
118from camcops_server.cc_modules.cc_xml import (
119 XML_COMMENT_SPECIAL_NOTES,
120 XmlElement,
121)
123if TYPE_CHECKING:
124 from sqlalchemy.sql.elements import ColumnElement
126 from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient
127 from camcops_server.cc_modules.cc_group import Group
128 from camcops_server.cc_modules.cc_policy import TokenizedPolicy
129 from camcops_server.cc_modules.cc_request import CamcopsRequest
130 from camcops_server.cc_modules.cc_taskschedule import PatientTaskSchedule
131 from camcops_server.cc_modules.cc_user import User
133log = BraceStyleAdapter(logging.getLogger(__name__))
136# =============================================================================
137# Patient class
138# =============================================================================
141class Patient(GenericTabletRecordMixin, Base):
142 """
143 Class representing a patient.
144 """
146 __tablename__ = "patient"
148 id: Mapped[int] = mapped_column(
149 comment="Primary key (patient ID) on the source tablet device",
150 # client PK
151 )
152 uuid: Mapped[Optional[python_uuid.UUID]] = mapped_camcops_column(
153 UuidColType,
154 comment="UUID",
155 default=python_uuid.uuid4, # generates a random UUID
156 )
157 forename: Mapped[Optional[str]] = mapped_camcops_column(
158 PatientNameColType,
159 index=True,
160 identifies_patient=True,
161 include_in_anon_staging_db=True,
162 comment="Forename",
163 )
164 surname: Mapped[Optional[str]] = mapped_camcops_column(
165 PatientNameColType,
166 index=True,
167 identifies_patient=True,
168 include_in_anon_staging_db=True,
169 comment="Surname",
170 )
171 dob: Mapped[Optional[datetime.date]] = mapped_camcops_column(
172 index=True,
173 identifies_patient=True,
174 include_in_anon_staging_db=True,
175 comment="Date of birth",
176 # ... e.g. "2013-02-04"
177 )
178 sex: Mapped[Optional[str]] = mapped_camcops_column(
179 SexColType,
180 index=True,
181 include_in_anon_staging_db=True,
182 comment="Sex (M, F, X)",
183 )
184 address: Mapped[Optional[str]] = mapped_camcops_column(
185 UnicodeText, identifies_patient=True, comment="Address"
186 )
187 email: Mapped[Optional[str]] = mapped_camcops_column(
188 EmailAddressColType,
189 identifies_patient=True,
190 comment="Patient's e-mail address",
191 )
192 gp: Mapped[Optional[str]] = mapped_camcops_column(
193 UnicodeText,
194 identifies_patient=True,
195 comment="General practitioner (GP)",
196 )
197 other: Mapped[Optional[str]] = mapped_camcops_column(
198 UnicodeText, identifies_patient=True, comment="Other details"
199 )
200 idnums: Mapped[list["PatientIdNum"]] = relationship(
201 # https://docs.sqlalchemy.org/en/latest/orm/join_conditions.html#relationship-custom-foreign
202 # https://docs.sqlalchemy.org/en/latest/orm/relationship_api.html#sqlalchemy.orm.relationship # noqa
203 # https://docs.sqlalchemy.org/en/latest/orm/join_conditions.html#relationship-primaryjoin # noqa
204 primaryjoin=(
205 "and_("
206 " remote(PatientIdNum.patient_id) == foreign(Patient.id), "
207 " remote(PatientIdNum._device_id) == foreign(Patient._device_id), "
208 " remote(PatientIdNum._era) == foreign(Patient._era), "
209 " remote(PatientIdNum._current) == True "
210 ")"
211 ),
212 uselist=True,
213 viewonly=True,
214 # Profiling results 2019-10-14 exporting 4185 phq9 records with
215 # unique patients to xlsx (task-patient relationship "selectin")
216 # lazy="select" : 35.3s
217 # lazy="joined" : 27.3s
218 # lazy="subquery": 15.2s (31.0s when task-patient also subquery)
219 # lazy="selectin": 26.4s
220 # See also patient relationship on Task class (cc_task.py)
221 lazy="subquery",
222 )
224 task_schedules: Mapped[list["PatientTaskSchedule"]] = relationship(
225 back_populates="patient",
226 cascade="all, delete",
227 cascade_backrefs=False,
228 )
230 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
231 # THE FOLLOWING ARE DEFUNCT, AND THE SERVER WORKS AROUND OLD TABLETS IN
232 # THE UPLOAD API.
233 #
234 # idnum1 = Column("idnum1", BigInteger, comment="ID number 1")
235 # idnum2 = Column("idnum2", BigInteger, comment="ID number 2")
236 # idnum3 = Column("idnum3", BigInteger, comment="ID number 3")
237 # idnum4 = Column("idnum4", BigInteger, comment="ID number 4")
238 # idnum5 = Column("idnum5", BigInteger, comment="ID number 5")
239 # idnum6 = Column("idnum6", BigInteger, comment="ID number 6")
240 # idnum7 = Column("idnum7", BigInteger, comment="ID number 7")
241 # idnum8 = Column("idnum8", BigInteger, comment="ID number 8")
242 #
243 # iddesc1 = Column("iddesc1", IdDescriptorColType, comment="ID description 1") # noqa
244 # iddesc2 = Column("iddesc2", IdDescriptorColType, comment="ID description 2") # noqa
245 # iddesc3 = Column("iddesc3", IdDescriptorColType, comment="ID description 3") # noqa
246 # iddesc4 = Column("iddesc4", IdDescriptorColType, comment="ID description 4") # noqa
247 # iddesc5 = Column("iddesc5", IdDescriptorColType, comment="ID description 5") # noqa
248 # iddesc6 = Column("iddesc6", IdDescriptorColType, comment="ID description 6") # noqa
249 # iddesc7 = Column("iddesc7", IdDescriptorColType, comment="ID description 7") # noqa
250 # iddesc8 = Column("iddesc8", IdDescriptorColType, comment="ID description 8") # noqa
251 #
252 # idshortdesc1 = Column("idshortdesc1", IdDescriptorColType, comment="ID short description 1") # noqa
253 # idshortdesc2 = Column("idshortdesc2", IdDescriptorColType, comment="ID short description 2") # noqa
254 # idshortdesc3 = Column("idshortdesc3", IdDescriptorColType, comment="ID short description 3") # noqa
255 # idshortdesc4 = Column("idshortdesc4", IdDescriptorColType, comment="ID short description 4") # noqa
256 # idshortdesc5 = Column("idshortdesc5", IdDescriptorColType, comment="ID short description 5") # noqa
257 # idshortdesc6 = Column("idshortdesc6", IdDescriptorColType, comment="ID short description 6") # noqa
258 # idshortdesc7 = Column("idshortdesc7", IdDescriptorColType, comment="ID short description 7") # noqa
259 # idshortdesc8 = Column("idshortdesc8", IdDescriptorColType, comment="ID short description 8") # noqa
260 #
261 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
263 # -------------------------------------------------------------------------
264 # Relationships
265 # -------------------------------------------------------------------------
267 # noinspection PyMethodParameters
268 @declared_attr # type: ignore[arg-type]
269 def special_notes(cls) -> RelationshipProperty:
270 """
271 Relationship to all :class:`SpecialNote` objects associated with this
272 patient.
273 """
274 # The SpecialNote also allows a link to patients, not just tasks,
275 # like this:
276 return relationship(
277 SpecialNote,
278 primaryjoin=(
279 "and_("
280 " remote(SpecialNote.basetable) == literal({repr_patient_tablename}), " # noqa
281 " remote(SpecialNote.task_id) == foreign(Patient.id), "
282 " remote(SpecialNote.device_id) == foreign(Patient._device_id), " # noqa
283 " remote(SpecialNote.era) == foreign(Patient._era), "
284 " not_(SpecialNote.hidden)"
285 ")".format(repr_patient_tablename=repr(cls.__tablename__))
286 ),
287 uselist=True,
288 order_by="SpecialNote.note_at",
289 viewonly=True, # for now!
290 )
292 # -------------------------------------------------------------------------
293 # Patient-fetching classmethods
294 # -------------------------------------------------------------------------
296 @classmethod
297 def get_patients_by_idnum(
298 cls,
299 dbsession: SqlASession,
300 which_idnum: int,
301 idnum_value: int,
302 group_id: int = None,
303 current_only: bool = True,
304 ) -> List["Patient"]:
305 """
306 Get all patients matching the specified ID number.
308 Args:
309 dbsession: a :class:`sqlalchemy.orm.session.Session`
310 which_idnum: which ID number type?
311 idnum_value: actual value of the ID number
312 group_id: optional group ID to restrict to
313 current_only: restrict to ``_current`` patients?
315 Returns:
316 list of all matching patients
318 """
319 if not which_idnum or which_idnum < 1:
320 return []
321 if idnum_value is None:
322 return []
323 q = dbsession.query(cls).join(cls.idnums) # type: ignore[arg-type]
324 # ... the join pre-restricts to current ID numbers
325 # https://docs.sqlalchemy.org/en/latest/orm/join_conditions.html#using-custom-operators-in-join-conditions # noqa
326 q = q.filter(PatientIdNum.which_idnum == which_idnum)
327 q = q.filter(PatientIdNum.idnum_value == idnum_value)
328 if group_id is not None:
329 q = q.filter(Patient._group_id == group_id)
330 if current_only:
331 q = q.filter(cls._current == True) # noqa: E712
332 patients = q.all() # type: List[Patient]
333 return patients
335 @classmethod
336 def get_patient_by_pk(
337 cls, dbsession: SqlASession, server_pk: int
338 ) -> Optional["Patient"]:
339 """
340 Fetch a patient by the server PK.
341 """
342 return dbsession.query(cls).filter(cls._pk == server_pk).first()
344 @classmethod
345 def get_patient_by_id_device_era(
346 cls, dbsession: SqlASession, client_id: int, device_id: int, era: str
347 ) -> Optional["Patient"]:
348 """
349 Fetch a patient by the client ID, device ID, and era.
350 """
351 return (
352 dbsession.query(cls)
353 .filter(cls.id == client_id)
354 .filter(cls._device_id == device_id)
355 .filter(cls._era == era)
356 .first()
357 )
359 # -------------------------------------------------------------------------
360 # String representations
361 # -------------------------------------------------------------------------
363 def __str__(self) -> str:
364 """
365 A plain string version, without the need for a request object.
367 Example:
369 .. code-block:: none
371 SMITH, BOB (M, 1 Jan 1950, idnum1=123, idnum2=456)
372 """
373 return "{sf} ({sex}, {dob}, {ids})".format(
374 sf=self.get_surname_forename_upper(),
375 sex=self.sex,
376 dob=self.get_dob_str(),
377 ids=", ".join(str(i) for i in self.get_idnum_objects()),
378 )
380 def prettystr(self, req: "CamcopsRequest") -> str:
381 """
382 A prettified string version.
384 Args:
385 req: a :class:`camcops_server.cc_modules.cc_request.CamcopsRequest`
387 Example:
389 .. code-block:: none
391 SMITH, BOB (M, 1 Jan 1950, RiO# 123, NHS# 456)
392 """
393 return "{sf} ({sex}, {dob}, {ids})".format(
394 sf=self.get_surname_forename_upper(),
395 sex=self.sex,
396 dob=self.get_dob_str(),
397 ids=", ".join(i.prettystr(req) for i in self.get_idnum_objects()),
398 )
400 def get_letter_style_identifiers(self, req: "CamcopsRequest") -> str:
401 """
402 Our best guess at the kind of text you'd put in a clinical letter to
403 say "it's about this patient".
405 Example:
407 .. code-block:: none
409 Bob Smith (1 Jan 1950, RiO number 123, NHS number 456)
410 """
411 return "{fs} ({dob}, {ids})".format(
412 fs=self.get_forename_surname(),
413 dob=self.get_dob_str(),
414 ids=", ".join(
415 i.full_prettystr(req) for i in self.get_idnum_objects()
416 ),
417 )
419 # -------------------------------------------------------------------------
420 # Equality
421 # -------------------------------------------------------------------------
423 def __eq__(self, other: object) -> bool:
424 """
425 Is this patient the same as another?
427 .. code-block:: python
429 from camcops_server.cc_modules.cc_patient import Patient
430 p1 = Patient(id=1, _device_id=1, _era="NOW")
431 print(p1 == p1) # True
432 p2 = Patient(id=1, _device_id=1, _era="NOW")
433 print(p1 == p2) # True
434 p3 = Patient(id=1, _device_id=2, _era="NOW")
435 print(p1 == p3) # False
437 s = set([p1, p2, p3]) # contains two patients
439 IMPERFECT in that it doesn't use intermediate patients to link
440 identity (e.g. P1 has RiO#=3, P2 has RiO#=3, NHS#=5, P3 has NHS#=5;
441 they are all the same by inference but P1 and P3 will not compare
442 equal).
444 """
445 # Same object?
446 # log.debug("self={}, other={}", self, other)
447 if self is other:
448 # log.debug("... same object; equal")
449 return True
451 if not isinstance(other, Patient):
452 # Since SQLAlchemy 2.0, when lazy-loading from related objects
453 # (e.g. Task.patient) the patient will be compared with
454 # non-patient SQLA internal status codes so we need to cater for
455 # this. It is probably good practice anyway.
457 # MyPy does not recognise try... except AttributeError
458 return NotImplemented
460 # Same device/era/patient ID (client PK)? Test int before str for speed
461 if (
462 self.id == other.id
463 and self._device_id == other._device_id
464 and self._era == other._era
465 and self.id is not None
466 and self._device_id is not None
467 and self._era is not None
468 ):
469 # log.debug("... same device/era/id; equal")
470 return True
471 # Shared ID number?
472 for sid in self.idnums:
473 if sid in other.idnums:
474 # log.debug("... share idnum {}; equal", sid)
475 return True
476 # Otherwise...
477 # log.debug("... unequal")
478 return False
480 def __hash__(self) -> int:
481 """
482 To put objects into a set, they must be hashable.
483 See https://docs.python.org/3/glossary.html#term-hashable.
484 If two objects are equal (via :func:`__eq__`) they must provide the
485 same hash value (but two objects with the same hash are not necessarily
486 equal).
487 """
488 return 0 # all objects have the same hash; "use __eq__() instead"
490 # -------------------------------------------------------------------------
491 # ID numbers
492 # -------------------------------------------------------------------------
494 def get_idnum_objects(self) -> List[PatientIdNum]:
495 """
496 Returns all :class:`PatientIdNum` objects for the patient.
498 These are SQLAlchemy ORM objects.
499 """
500 return self.idnums
502 def get_idnum_references(self) -> List[IdNumReference]:
503 """
504 Returns all
505 :class:`camcops_server.cc_modules.cc_simpleobjects.IdNumReference`
506 objects for the patient.
508 These are simple which_idnum/idnum_value pairs.
509 """
510 idnums = self.idnums # type: List[PatientIdNum]
511 return [
512 x.get_idnum_reference()
513 for x in idnums
514 if x.is_superficially_valid()
515 ]
517 def get_idnum_raw_values_only(self) -> List[int]:
518 """
519 Get all plain ID number values (ignoring which ID number type they
520 represent) for the patient.
521 """
522 idnums = self.idnums # type: List[PatientIdNum]
523 return [x.idnum_value for x in idnums if x.is_superficially_valid()]
525 def get_idnum_object(self, which_idnum: int) -> Optional[PatientIdNum]:
526 """
527 Gets the PatientIdNum object for a specified which_idnum, or None.
528 """
529 idnums = self.idnums # type: List[PatientIdNum]
530 for x in idnums:
531 if x.which_idnum == which_idnum:
532 return x
533 return None
535 def has_idnum_type(self, which_idnum: int) -> bool:
536 """
537 Does the patient have an ID number of the specified type?
538 """
539 return self.get_idnum_object(which_idnum) is not None
541 def get_idnum_value(self, which_idnum: int) -> Optional[int]:
542 """
543 Get value of a specific ID number, if present.
544 """
545 idobj = self.get_idnum_object(which_idnum)
546 return idobj.idnum_value if idobj else None
548 def set_idnum_value(
549 self, req: "CamcopsRequest", which_idnum: int, idnum_value: int
550 ) -> None:
551 """
552 Sets an ID number value.
553 """
554 dbsession = req.dbsession
555 ccsession = req.camcops_session
556 idnums = self.idnums # type: List[PatientIdNum]
557 for idobj in idnums:
558 if idobj.which_idnum == which_idnum:
559 idobj.idnum_value = idnum_value
560 return
561 # Otherwise, make a new one:
562 newid = PatientIdNum()
563 newid.patient_id = self.id
564 newid._device_id = self._device_id
565 newid._era = self._era
566 newid._current = True
567 newid._when_added_exact = req.now_era_format
568 newid._when_added_batch_utc = req.now_utc
569 newid._adding_user_id = ccsession.user_id
570 newid._camcops_version = CAMCOPS_SERVER_VERSION_STRING
571 dbsession.add(newid)
572 self.idnums.append(newid)
574 def get_iddesc(
575 self, req: "CamcopsRequest", which_idnum: int
576 ) -> Optional[str]:
577 """
578 Get value of a specific ID description, if present.
579 """
580 idobj = self.get_idnum_object(which_idnum)
581 return idobj.description(req) if idobj else None
583 def get_idshortdesc(
584 self, req: "CamcopsRequest", which_idnum: int
585 ) -> Optional[str]:
586 """
587 Get value of a specific ID short description, if present.
588 """
589 idobj = self.get_idnum_object(which_idnum)
590 return idobj.short_description(req) if idobj else None
592 def add_extra_idnum_info_to_row(self, row: Dict[str, Any]) -> None:
593 """
594 For the ``DB_PATIENT_ID_PER_ROW`` export option. Adds additional ID
595 number info to a row.
597 Args:
598 row: future database row, as a dictionary
599 """
600 for idobj in self.idnums:
601 which_idnum = idobj.which_idnum
602 fieldname = extra_id_colname(which_idnum)
603 row[fieldname] = idobj.idnum_value
605 # -------------------------------------------------------------------------
606 # Group
607 # -------------------------------------------------------------------------
609 @property
610 def group(self) -> Optional["Group"]:
611 """
612 Returns the :class:`camcops_server.cc_modules.cc_group.Group` to which
613 this patient's record belongs.
614 """
615 return self._group
617 # -------------------------------------------------------------------------
618 # Policies
619 # -------------------------------------------------------------------------
621 def satisfies_upload_id_policy(self) -> bool:
622 """
623 Does the patient satisfy the uploading ID policy?
624 """
625 group = self._group # type: Optional[Group]
626 if not group:
627 return False
628 return self.satisfies_id_policy(group.tokenized_upload_policy())
630 def satisfies_finalize_id_policy(self) -> bool:
631 """
632 Does the patient satisfy the finalizing ID policy?
633 """
634 group = self._group # type: Optional[Group]
635 if not group:
636 return False
637 return self.satisfies_id_policy(group.tokenized_finalize_policy())
639 def satisfies_id_policy(self, policy: "TokenizedPolicy") -> bool:
640 """
641 Does the patient satisfy a particular ID policy?
642 """
643 return policy.satisfies_id_policy(self.get_bare_ptinfo())
645 # -------------------------------------------------------------------------
646 # Name, DOB/age, sex, address, etc.
647 # -------------------------------------------------------------------------
649 def get_surname(self) -> str:
650 """
651 Get surname (in upper case) or "".
652 """
653 return self.surname.upper() if self.surname else ""
655 def get_forename(self) -> str:
656 """
657 Get forename (in upper case) or "".
658 """
659 return self.forename.upper() if self.forename else ""
661 def get_forename_surname(self) -> str:
662 """
663 Get "Forename Surname" as a string, using "(UNKNOWN)" for missing
664 details.
665 """
666 f = self.forename or "(UNKNOWN)"
667 s = self.surname or "(UNKNOWN)"
668 return f"{f} {s}"
670 def get_surname_forename_upper(self) -> str:
671 """
672 Get "SURNAME, FORENAME", using "(UNKNOWN)" for missing details.
673 """
674 s = self.surname.upper() if self.surname else "(UNKNOWN)"
675 f = self.forename.upper() if self.forename else "(UNKNOWN)"
676 return f"{s}, {f}"
678 def get_dob_html(self, req: "CamcopsRequest", longform: bool) -> str:
679 """
680 HTML fragment for date of birth.
681 """
682 _ = req.gettext
683 if longform:
684 dob = answer(
685 format_datetime(self.dob, DateFormat.LONG_DATE, default=None)
686 )
688 dobtext = _("Date of birth:")
689 return f"<br>{dobtext} {dob}"
690 else:
691 dobtext = _("DOB:")
692 dob = format_datetime(self.dob, DateFormat.SHORT_DATE)
693 return f"{dobtext} {dob}."
695 def get_age(
696 self, req: "CamcopsRequest", default: str = ""
697 ) -> Union[int, str]:
698 """
699 Age (in whole years) today, or default.
700 """
701 now = req.now
702 return self.get_age_at(now, default=default)
704 def get_dob(self) -> Optional[pendulum.Date]:
705 """
706 Date of birth, as a a timezone-naive date.
707 """
708 dob = self.dob
709 if not dob:
710 return None
711 return coerce_to_pendulum_date(dob)
713 def get_dob_str(self) -> Optional[str]:
714 """
715 Date of birth, as a string.
716 """
717 dob_dt = self.get_dob()
718 if dob_dt is None:
719 return None
720 return format_datetime(dob_dt, DateFormat.SHORT_DATE)
722 def get_age_at(
723 self, when: PotentialDatetimeType, default: str = ""
724 ) -> Union[int, str]:
725 """
726 Age (in whole years) at a particular date, or default.
727 """
728 return get_age(self.dob, when, default=default)
730 def is_female(self) -> bool:
731 """
732 Is sex 'F'?
733 """
734 return self.sex == SEX_FEMALE
736 def is_male(self) -> bool:
737 """
738 Is sex 'M'?
739 """
740 return self.sex == SEX_MALE
742 def get_sex(self) -> str:
743 """
744 Return sex or "".
745 """
746 return self.sex or ""
748 def get_sex_verbose(self, default: str = "sex unknown") -> str:
749 """
750 Returns HTML-safe version of sex, or default.
751 """
752 return default if not self.sex else ws.webify(self.sex)
754 def get_address(self) -> Optional[str]:
755 """
756 Returns address (NOT necessarily web-safe).
757 """
758 address = self.address # type: Optional[str]
759 return address or ""
761 def get_email(self) -> Optional[str]:
762 """
763 Returns email address
764 """
765 email = self.email # type: Optional[str]
766 return email or ""
768 # -------------------------------------------------------------------------
769 # Other representations
770 # -------------------------------------------------------------------------
772 def get_xml_root(
773 self, req: "CamcopsRequest", options: TaskExportOptions = None
774 ) -> XmlElement:
775 """
776 Get root of XML tree, as an
777 :class:`camcops_server.cc_modules.cc_xml.XmlElement`.
779 Args:
780 req: a :class:`camcops_server.cc_modules.cc_request.CamcopsRequest`
781 options: a :class:`camcops_server.cc_modules.cc_simpleobjects.TaskExportOptions`
782 """ # noqa
783 # No point in skipping old ID columns (1-8) now; they're gone.
784 branches = self._get_xml_branches(req, options=options)
785 # Now add new-style IDs:
786 pidnum_branches = [] # type: List[XmlElement]
787 pidnum_options = TaskExportOptions(
788 xml_include_plain_columns=True, xml_with_header_comments=False
789 )
790 for pidnum in self.idnums: # type: PatientIdNum
791 pidnum_branches.append(
792 pidnum._get_xml_root(req, options=pidnum_options)
793 )
794 branches.append(XmlElement(name="idnums", value=pidnum_branches))
795 # Special notes
796 branches.append(XML_COMMENT_SPECIAL_NOTES)
797 special_notes = self.special_notes # type: List[SpecialNote]
798 for sn in special_notes:
799 branches.append(sn.get_xml_root())
800 return XmlElement(name=self.__tablename__, value=branches)
802 def get_spreadsheet_page(self, req: "CamcopsRequest") -> SpreadsheetPage:
803 """
804 Get a :class:`camcops_server.cc_modules.cc_spreadsheet.SpreadsheetPage`
805 for the patient.
806 """
807 # 1. Our core fields.
808 page = self._get_core_spreadsheet_page(
809 req, heading_prefix=SPREADSHEET_PATIENT_FIELD_PREFIX
810 )
811 # 2. ID number details
812 # We can't just iterate through the ID numbers; we have to iterate
813 # through all possible ID numbers.
814 for iddef in req.idnum_definitions:
815 n = iddef.which_idnum
816 nstr = str(n)
817 shortdesc = iddef.short_description
818 longdesc = iddef.description
819 idnum_value = next(
820 (
821 idnum.idnum_value
822 for idnum in self.idnums
823 if idnum.which_idnum == n
824 and idnum.is_superficially_valid()
825 ),
826 None,
827 )
828 page.add_or_set_value(
829 heading=SPREADSHEET_PATIENT_FIELD_PREFIX + FP_ID_NUM + nstr,
830 value=idnum_value,
831 )
832 page.add_or_set_value(
833 heading=SPREADSHEET_PATIENT_FIELD_PREFIX + FP_ID_DESC + nstr,
834 value=longdesc,
835 )
836 page.add_or_set_value(
837 heading=(
838 SPREADSHEET_PATIENT_FIELD_PREFIX + FP_ID_SHORT_DESC + nstr
839 ),
840 value=shortdesc,
841 )
842 return page
844 def get_spreadsheet_schema_elements(
845 self, req: "CamcopsRequest", table_name: str = ""
846 ) -> Set[SummarySchemaInfo]:
847 """
848 Follows :func:`get_spreadsheet_page`, but retrieving schema
849 information.
850 """
851 # 1. Core fields
852 items = self._get_core_spreadsheet_schema(
853 table_name=table_name,
854 column_name_prefix=SPREADSHEET_PATIENT_FIELD_PREFIX,
855 )
856 # 2. ID number details
857 table_name = table_name or self.__tablename__
858 for iddef in req.idnum_definitions:
859 n = iddef.which_idnum
860 nstr = str(n)
861 comment_suffix = f" [ID#{n}]"
862 items.add(
863 SummarySchemaInfo(
864 table_name=table_name,
865 source=SummarySchemaInfo.SSV_DB,
866 column_name=(
867 SPREADSHEET_PATIENT_FIELD_PREFIX + FP_ID_NUM + nstr
868 ),
869 data_type=str(PatientIdNum.idnum_value.type),
870 comment=PatientIdNum.idnum_value.comment + comment_suffix,
871 )
872 )
873 items.add(
874 SummarySchemaInfo(
875 table_name=table_name,
876 source=SummarySchemaInfo.SSV_DB,
877 column_name=(
878 SPREADSHEET_PATIENT_FIELD_PREFIX + FP_ID_DESC + nstr
879 ),
880 data_type=str(IdNumDefinition.description.type),
881 comment=IdNumDefinition.description.comment
882 + comment_suffix,
883 )
884 )
885 items.add(
886 SummarySchemaInfo(
887 table_name=table_name,
888 source=SummarySchemaInfo.SSV_DB,
889 column_name=(
890 SPREADSHEET_PATIENT_FIELD_PREFIX
891 + FP_ID_SHORT_DESC
892 + nstr
893 ),
894 data_type=str(IdNumDefinition.short_description.type),
895 comment=(
896 IdNumDefinition.short_description.comment
897 + comment_suffix
898 ),
899 )
900 )
901 return items
903 def get_bare_ptinfo(self) -> BarePatientInfo:
904 """
905 Get basic identifying information, as a
906 :class:`camcops_server.cc_modules.cc_simpleobjects.BarePatientInfo`
907 object.
908 """
909 return BarePatientInfo(
910 forename=self.forename,
911 surname=self.surname,
912 sex=self.sex,
913 dob=self.dob, # type: ignore[arg-type]
914 address=self.address,
915 email=self.email,
916 gp=self.gp,
917 otherdetails=self.other,
918 idnum_definitions=self.get_idnum_references(),
919 )
921 def get_hl7_pid_segment(
922 self, req: "CamcopsRequest", recipient: "ExportRecipient"
923 ) -> hl7.Segment:
924 """
925 Get HL7 patient identifier (PID) segment.
927 Args:
928 req:
929 a :class:`camcops_server.cc_modules.cc_request.CamcopsRequest`
930 recipient:
931 a :class:`camcops_server.cc_modules.cc_exportrecipient.ExportRecipient`
933 Returns:
934 a :class:`hl7.Segment` object
935 """ # noqa
936 # Put the primary one first:
937 patient_id_tuple_list = [
938 HL7PatientIdentifier(
939 pid=str(self.get_idnum_value(recipient.primary_idnum)),
940 id_type=recipient.get_hl7_id_type(
941 req, recipient.primary_idnum
942 ),
943 assigning_authority=recipient.get_hl7_id_aa(
944 req, recipient.primary_idnum
945 ),
946 )
947 ]
948 # Then the rest:
949 for idobj in self.idnums:
950 which_idnum = idobj.which_idnum
951 if which_idnum == recipient.primary_idnum:
952 continue
953 idnum_value = idobj.idnum_value
954 if idnum_value is None:
955 continue
956 patient_id_tuple_list.append(
957 HL7PatientIdentifier(
958 pid=str(idnum_value),
959 id_type=recipient.get_hl7_id_type(req, which_idnum),
960 assigning_authority=recipient.get_hl7_id_aa(
961 req, which_idnum
962 ),
963 )
964 )
965 return make_pid_segment(
966 forename=self.get_surname(),
967 surname=self.get_forename(),
968 dob=self.get_dob(),
969 sex=self.get_sex(),
970 address=self.get_address(),
971 patient_id_list=patient_id_tuple_list,
972 )
974 # -------------------------------------------------------------------------
975 # FHIR
976 # -------------------------------------------------------------------------
978 def get_fhir_bundle_entry(
979 self, req: "CamcopsRequest", recipient: "ExportRecipient"
980 ) -> Dict[str, Any]:
981 """
982 Returns a dictionary, suitable for serializing to JSON, that
983 encapsulates patient identity information in a FHIR bundle.
985 See https://www.hl7.org/fhir/patient.html.
986 """
987 # The JSON objects we will build up:
988 patient_dict = {} # type: JsonObjectType
990 # Name
991 if self.forename or self.surname:
992 name_dict = {} # type: JsonObjectType
993 if self.forename:
994 name_dict[Fc.NAME_GIVEN] = [self.forename]
995 if self.surname:
996 name_dict[Fc.NAME_FAMILY] = self.surname
997 patient_dict[Fc.NAME] = [HumanName(jsondict=name_dict).as_json()]
999 # DOB
1000 if self.dob:
1001 patient_dict[Fc.BIRTHDATE] = format_datetime(
1002 self.dob, DateFormat.FILENAME_DATE_ONLY
1003 )
1005 # Sex/gender (should always be present, per client minimum ID policy)
1006 if self.sex:
1007 gender_lookup = {
1008 SEX_FEMALE: Fc.GENDER_FEMALE,
1009 SEX_MALE: Fc.GENDER_MALE,
1010 SEX_OTHER_UNSPECIFIED: Fc.GENDER_OTHER,
1011 }
1012 patient_dict[Fc.GENDER] = gender_lookup.get(
1013 self.sex, Fc.GENDER_UNKNOWN
1014 )
1016 # Address
1017 if self.address:
1018 patient_dict[Fc.ADDRESS] = [
1019 Address(jsondict={Fc.ADDRESS_TEXT: self.address}).as_json()
1020 ]
1022 # Email
1023 if self.email:
1024 patient_dict[Fc.TELECOM] = [
1025 ContactPoint(
1026 jsondict={
1027 Fc.SYSTEM: Fc.TELECOM_SYSTEM_EMAIL,
1028 Fc.VALUE: self.email,
1029 }
1030 ).as_json()
1031 ]
1033 # General practitioner (GP): via
1034 # fhirclient.models.fhirreference.FHIRReference; too structured.
1036 # ID numbers go here:
1037 return make_fhir_bundle_entry(
1038 resource_type_url=Fc.RESOURCE_TYPE_PATIENT,
1039 identifier=self.get_fhir_identifier(req, recipient),
1040 resource=FhirPatient(jsondict=patient_dict).as_json(),
1041 )
1043 def get_fhir_identifier(
1044 self, req: "CamcopsRequest", recipient: "ExportRecipient"
1045 ) -> Identifier:
1046 """
1047 Returns a FHIR identifier for this patient, as a
1048 :class:`fhirclient.models.identifier.Identifier` object.
1050 This pairs a URL to our CamCOPS server indicating the ID number type
1051 (as the "system") with the actual ID number (as the "value").
1053 For debugging situations, it falls back to a default identifier (using
1054 the PK on our CamCOPS server).
1055 """
1056 which_idnum = recipient.primary_idnum
1057 try:
1058 # For real exports, the fact that the patient does have an ID
1059 # number of the right type will have been pre-verified.
1060 if which_idnum is None:
1061 raise AttributeError
1062 idnum_object = self.get_idnum_object(which_idnum)
1063 idnum_value = idnum_object.idnum_value # may raise AttributeError
1064 iddef = req.get_idnum_definition(which_idnum)
1065 idnum_url = iddef.effective_fhir_id_system(req)
1066 return Identifier(
1067 jsondict={Fc.SYSTEM: idnum_url, Fc.VALUE: str(idnum_value)}
1068 )
1069 except AttributeError:
1070 # We are probably in a debugging/drafting situation. Fall back to
1071 # a default identifier.
1072 return fhir_pk_identifier(
1073 req,
1074 self.__tablename__,
1075 self.pk,
1076 Fc.CAMCOPS_VALUE_PATIENT_WITHIN_TASK,
1077 )
1079 def get_fhir_subject_ref(
1080 self, req: "CamcopsRequest", recipient: "ExportRecipient"
1081 ) -> Dict:
1082 """
1083 Returns a FHIRReference (in JSON dict format) used to refer to this
1084 patient as a "subject" of some other entry (like a questionnaire).
1085 """
1086 return FHIRReference(
1087 jsondict={
1088 Fc.TYPE: Fc.RESOURCE_TYPE_PATIENT,
1089 Fc.IDENTIFIER: self.get_fhir_identifier(
1090 req, recipient
1091 ).as_json(),
1092 }
1093 ).as_json()
1095 # -------------------------------------------------------------------------
1096 # Database status
1097 # -------------------------------------------------------------------------
1099 def is_preserved(self) -> bool:
1100 """
1101 Is the patient record preserved and erased from the tablet?
1102 """
1103 return self._pk is not None and self._era != ERA_NOW
1105 # -------------------------------------------------------------------------
1106 # Audit
1107 # -------------------------------------------------------------------------
1109 def audit(
1110 self, req: "CamcopsRequest", details: str, from_console: bool = False
1111 ) -> None:
1112 """
1113 Audits an action to this patient.
1114 """
1115 audit(
1116 req,
1117 details,
1118 patient_server_pk=self._pk,
1119 table=Patient.__tablename__,
1120 server_pk=self._pk,
1121 from_console=from_console,
1122 )
1124 # -------------------------------------------------------------------------
1125 # Special notes
1126 # -------------------------------------------------------------------------
1128 def apply_special_note(
1129 self,
1130 req: "CamcopsRequest",
1131 note: str,
1132 audit_msg: str = "Special note applied manually",
1133 ) -> None:
1134 """
1135 Manually applies a special note to a patient.
1136 WRITES TO DATABASE.
1137 """
1138 sn = SpecialNote()
1139 sn.basetable = self.__tablename__
1140 sn.task_id = self.id # patient ID, in this case
1141 sn.device_id = self._device_id
1142 sn.era = self._era
1143 sn.note_at = req.now
1144 sn.user_id = req.user_id
1145 sn.note = note
1146 req.dbsession.add(sn)
1147 self.special_notes.append(sn) # type: ignore[attr-defined]
1148 self.audit(req, audit_msg)
1149 # HL7 deletion of corresponding tasks is done in camcops_server.py
1151 # -------------------------------------------------------------------------
1152 # Deletion
1153 # -------------------------------------------------------------------------
1155 def gen_patient_idnums_even_noncurrent(
1156 self,
1157 ) -> Generator[PatientIdNum, None, None]:
1158 """
1159 Generates all :class:`PatientIdNum` objects, including non-current
1160 ones.
1161 """
1162 for lineage_member in self._gen_unique_lineage_objects( # type: ignore[assignment] # noqa: E501
1163 self.idnums
1164 ): # type: PatientIdNum
1165 yield lineage_member
1167 def delete_with_dependants(self, req: "CamcopsRequest") -> None:
1168 """
1169 Delete the patient with all its dependent objects.
1170 """
1171 if self._pk is None:
1172 return
1173 for pidnum in self.gen_patient_idnums_even_noncurrent():
1174 req.dbsession.delete(pidnum)
1175 super().delete_with_dependants(req)
1177 # -------------------------------------------------------------------------
1178 # Permissions
1179 # -------------------------------------------------------------------------
1181 def user_may_view(self, user: "User") -> bool:
1182 """
1183 May this user inspect patient details directly?
1184 """
1185 return self._group_id in user.ids_of_groups_user_may_see
1187 def user_may_edit(self, req: "CamcopsRequest") -> bool:
1188 """
1189 Does the current user have permission to edit this patient?
1190 """
1191 if self.created_on_server(req):
1192 # Anyone in the group with the right permission
1193 return req.user.may_manage_patients_in_group(self._group_id)
1195 # Finalized patient: Need to be group administrator
1196 return req.user.may_administer_group(self._group_id)
1198 # --------------------------------------------------------------------------
1199 # UUID
1200 # --------------------------------------------------------------------------
1201 @property
1202 def uuid_as_proquint(self) -> Optional[str]:
1203 # Convert integer into pronounceable quintuplets (proquint)
1204 # https://arxiv.org/html/0901.4016
1205 if self.uuid is None:
1206 return None
1208 return proquint_from_uuid(self.uuid)
1210 @property
1211 def duplicates(self) -> set["Patient"]:
1212 """
1213 Returns a set of patients that have an ID Number that matches this one.
1214 For a patient to be considered a duplicate, the ID number needs to
1215 match on:
1217 * Group
1218 * Type of ID Number (e.g. NHS Number)
1219 * Value of ID number
1220 * Device where patient was created
1222 and have the "current" flag set to True
1223 """
1225 dups = set()
1227 for idnum in self.idnums:
1228 for dup in idnum.duplicates:
1229 dups.add(dup.patient)
1231 return dups
1234# =============================================================================
1235# Validate candidate patient info for upload
1236# =============================================================================
1239def is_candidate_patient_valid_for_group(
1240 ptinfo: BarePatientInfo, group: "Group", finalizing: bool
1241) -> Tuple[bool, str]:
1242 """
1243 Is the specified patient acceptable to upload into this group?
1245 Checks:
1247 - group upload or finalize policy
1249 .. todo:: is_candidate_patient_valid: check against predefined patients, if
1250 the group wants
1252 Args:
1253 ptinfo:
1254 a
1255 :class:`camcops_server.cc_modules.cc_simpleobjects.BarePatientInfo`
1256 representing the patient info to check
1257 group:
1258 the :class:`camcops_server.cc_modules.cc_group.Group` into which
1259 this patient will be uploaded, if allowed
1260 finalizing:
1261 finalizing, rather than uploading?
1263 Returns:
1264 tuple: valid, reason
1266 """
1267 if not group:
1268 return False, "Nonexistent group"
1270 if finalizing:
1271 if not group.tokenized_finalize_policy().satisfies_id_policy(ptinfo):
1272 return False, "Fails finalizing ID policy"
1273 else:
1274 if not group.tokenized_upload_policy().satisfies_id_policy(ptinfo):
1275 return False, "Fails upload ID policy"
1277 # todo: add checks against prevalidated patients here
1279 return True, ""
1282def is_candidate_patient_valid_for_restricted_user(
1283 req: "CamcopsRequest", ptinfo: BarePatientInfo
1284) -> Tuple[bool, str]:
1285 """
1286 Is the specified patient OK to be uploaded by this user? Performs a check
1287 for restricted (single-patient) users; if true, ensures that the
1288 identifiers all match the expected patient.
1290 Args:
1291 req:
1292 the :class:`camcops_server.cc_modules.cc_request.CamcopsRequest`
1293 ptinfo:
1294 a
1295 :class:`camcops_server.cc_modules.cc_simpleobjects.BarePatientInfo`
1296 representing the patient info to check
1298 Returns:
1299 tuple: valid, reason
1300 """
1301 user = req.user
1302 if not user.auto_generated:
1303 # Not a restricted user; no problem.
1304 return True, ""
1306 server_patient = user.single_patient
1307 if not server_patient:
1308 return (
1309 False,
1310 (
1311 f"Restricted user {user.username} does not have associated "
1312 f"patient details"
1313 ),
1314 )
1316 server_ptinfo = server_patient.get_bare_ptinfo()
1317 if ptinfo != server_ptinfo:
1318 return False, f"Should be {server_ptinfo}"
1320 return True, ""
1323# =============================================================================
1324# Reports
1325# =============================================================================
1328class DistinctPatientReport(Report):
1329 """
1330 Report to show distinct patients.
1331 """
1333 # noinspection PyMethodParameters
1334 @classproperty
1335 def report_id(cls) -> str:
1336 return "patient_distinct"
1338 @classmethod
1339 def title(cls, req: "CamcopsRequest") -> str:
1340 _ = req.gettext
1341 return _(
1342 "(Server) Patients, distinct by name, sex, DOB, all ID " "numbers"
1343 )
1345 # noinspection PyMethodParameters
1346 @classproperty
1347 def superuser_only(cls) -> bool:
1348 return False
1350 # noinspection PyProtectedMember
1351 def get_query(self, req: "CamcopsRequest") -> SelectBase:
1352 select_fields: list[ColumnElement[Any]] = [
1353 Patient.surname.label("surname"),
1354 Patient.forename.label("forename"),
1355 Patient.dob.label("dob"),
1356 Patient.sex.label("sex"),
1357 ]
1358 # noinspection PyUnresolvedReferences
1359 select_from = Patient.__table__
1360 wheres = [
1361 Patient._current == True # noqa: E712
1362 ] # type: List[ClauseElement]
1363 if not req.user.superuser:
1364 # Restrict to accessible groups
1365 group_ids = req.user.ids_of_groups_user_may_report_on
1366 wheres.append(Patient._group_id.in_(group_ids))
1367 for iddef in req.idnum_definitions:
1368 n = iddef.which_idnum
1369 desc = iddef.short_description
1370 # noinspection PyUnresolvedReferences
1371 aliased_table = PatientIdNum.__table__.alias(f"i{n}")
1372 select_fields.append(aliased_table.c.idnum_value.label(desc))
1373 select_from = select_from.outerjoin(
1374 aliased_table,
1375 and_(
1376 aliased_table.c.patient_id == Patient.id,
1377 aliased_table.c._device_id == Patient._device_id,
1378 aliased_table.c._era == Patient._era,
1379 # Note: the following are part of the JOIN, not the WHERE:
1380 # (or failure to match a row will wipe out the Patient from
1381 # the OUTER JOIN):
1382 aliased_table.c._current == True, # noqa: E712
1383 aliased_table.c.which_idnum == n,
1384 ),
1385 ) # nopep8
1386 order_by = [
1387 Patient.surname,
1388 Patient.forename,
1389 Patient.dob,
1390 Patient.sex,
1391 ]
1392 query = (
1393 select(*select_fields)
1394 .select_from(select_from)
1395 .where(and_(*wheres)) # type: ignore[arg-type]
1396 .order_by(*order_by)
1397 .distinct()
1398 )
1399 return query