Coverage for crateweb/consent/lookup_systmone.py: 14%
184 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-08-27 10:34 -0500
1"""
2crate_anon/crateweb/consent/lookup_systmone.py
4===============================================================================
6 Copyright (C) 2015, University of Cambridge, Department of Psychiatry.
7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).
9 This file is part of CRATE.
11 CRATE 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 CRATE 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 CRATE. If not, see <https://www.gnu.org/licenses/>.
24===============================================================================
26**Functions to look up patient details from a TPP SystmOne Strategic Reporting
27Extract (SRE) database, or at least its CPFT equivalent.**
29"""
31import datetime
32from typing import Generator, List, Optional, Tuple
34from cardinal_pythonlib.dbfunc import dictfetchall, dictfetchone, genrows
35from cardinal_pythonlib.typing_helpers import (
36 Pep249DatabaseCursorType as Cursor,
37)
38from django.db import connections
40from crate_anon.crateweb.config.constants import ClinicalDatabaseType
41from crate_anon.crateweb.consent.lookup_common import (
42 get_team_details,
43 pick_best_clinician,
44 SignatoryTitles,
45)
46from crate_anon.crateweb.consent.models import (
47 ClinicianInfoHolder,
48 ConsentMode,
49 PatientLookup,
50)
51from crate_anon.crateweb.consent.utils import to_date
52from crate_anon.preprocess.systmone_ddgen import (
53 cpft_s1_tablename,
54 CPFTTable,
55 S1Table,
56)
59# =============================================================================
60# Constants
61# =============================================================================
63_GENDER_S1_TO_CRATE = {
64 "F": PatientLookup.FEMALE,
65 "I": PatientLookup.INTERSEX,
66 "M": PatientLookup.MALE,
67 "U": PatientLookup.UNKNOWNSEX,
68}
71# =============================================================================
72# CPFT staff details
73# =============================================================================
76def _is_staff_title(x: str) -> bool:
77 """
78 Does this string look like a title?
79 """
80 x = x.strip().rstrip(".").upper()
81 return x in ("DR", "MISS", "MR", "MRS", "MS", "PROF")
84def _is_initial(x: str) -> bool:
85 """
86 Does this bit look like an initial?
87 """
88 x = x.strip().rstrip(".")
89 return len(x) == 1
92def _get_staff_title_forename_surname(_combined: str) -> Tuple[str, str, str]:
93 """
94 Parse forename/surname for staff, in simple fashion. Generally, these are
95 either "Dr Alice Smith" or "Alice Smith".
96 """
97 parts = (_combined or "").split()
98 title = ""
99 if len(parts) > 1 and _is_staff_title(parts[0]):
100 title = parts[0]
101 parts = parts[1:]
102 fname = "".join(parts[:1])
103 parts = parts[1:]
104 if len(parts) > 1 and _is_initial(parts[0]):
105 # Skip initial
106 parts = parts[1:]
107 sname = " ".join(parts)
108 return title, fname, sname
111def _get_staff_details(
112 cursor: Cursor,
113 clinician_type: str,
114 signatory_title: str,
115 is_consultant: bool,
116 name: str = None,
117 profile_id: int = None,
118 start_date: datetime.date = None,
119 end_date: datetime.date = None,
120) -> Optional[ClinicianInfoHolder]:
121 """
122 Look up details about a member of staff, as best we can.
124 MAY RE-USE THE DATABASE CURSOR; the calling code needs to be happy about
125 that. [Note that our `dictfetchall()` function fetches everything in one
126 go, so that's OK -- the cursor is not going to be re-used later in a loop
127 -- and dictfetchone() is used in this file's code on a one-off basis only.]
129 Relevant database structure:
131 - Some tables have IDProfile<something_e.g._StaffMember>, which is a
132 foreign key to SRStaffMemberProfile.RowIdentifier). That seems the most
133 prevalent. Find these in CPFT with e.g.:
135 .. code-block:: sql
137 SELECT * FROM information_schema.columns
138 WHERE table_catalog = 'SystmOne' AND column_name LIKE '%IDProfile%';
140 - Some have StaffName, which is inserted by CPFT, likely from
141 SRStaffMember.StaffName.
143 - A few (e.g. CPFT's S1_ReferralAllocationStaff) have IDStaffMember, not
144 in the original. This is likely from SRStaffMemberProfile.IDStaffMember,
145 itself a key to SRStaffMember.RowIdentifier. SRStaffMember contains
146 names.
148 - So, the original TPP SRE uses IDProfile*, and the others have been added
149 by CPFT.
151 - However, staff e-mail addresses aren't obviously present anywhere.
153 - Also, as of 2023-10-25, the staff lookup tables are missing too.
155 """
156 # Look up from profile ID.
157 if profile_id:
158 assert cursor is not None # temporary, removes unused var warning
159 pass # todo: When SystmOne staff data available, implement lookup
160 # (a) fetch profile record from SRStaffMemberProfile
161 # (b) fetch name from SRStaffMember
163 # Poor version: name alone.
164 if name:
165 title, first_name, surname = _get_staff_title_forename_surname(name)
166 return ClinicianInfoHolder(
167 clinician_type=clinician_type,
168 title=title,
169 first_name=first_name,
170 surname=surname,
171 email="", # PROBLEM!
172 signatory_title=signatory_title,
173 is_consultant=is_consultant,
174 start_date=start_date,
175 end_date=end_date,
176 )
178 # Failed
179 return None
182def _process_consultant(
183 clinicians: List[ClinicianInfoHolder],
184 cursor: Cursor,
185 name: str = "",
186 profile_id: int = None,
187 start_date: datetime.date = None,
188 end_date: datetime.date = None,
189) -> None:
190 """
191 Look up details for a consultant and add them to "clinicians".
192 """
193 if profile_id is None:
194 return
195 consultant = _get_staff_details(
196 cursor=cursor,
197 clinician_type=ClinicianInfoHolder.CONSULTANT,
198 signatory_title=SignatoryTitles.CONSULTANT,
199 # ... not necessarily (though often) a psychiatrist; might be e.g. a
200 # consultant geriatrician.
201 is_consultant=True,
202 name=name,
203 profile_id=profile_id,
204 start_date=start_date,
205 end_date=end_date,
206 )
207 if not consultant:
208 return
209 clinicians.append(consultant)
212# =============================================================================
213# Look up patient IDs
214# =============================================================================
217def lookup_cpft_systmone(
218 lookup: PatientLookup, decisions: List[str], secret_decisions: List[str]
219) -> None:
220 """
221 Look up patient details from a TPP SystmOne Strategic Reporting Extract
222 (SRE) database.
224 Args:
225 lookup: a :class:`crate_anon.crateweb.consent.models.PatientLookup`
226 decisions: list of human-readable decisions; will be modified
227 secret_decisions: list of human-readable decisions containing secret
228 (identifiable) information; will be modified
229 """
230 cursor = connections[ClinicalDatabaseType.CPFT_SYSTMONE].cursor()
232 # -------------------------------------------------------------------------
233 # 1. Name, DOB, DOD, gender, e-mail address
234 # -------------------------------------------------------------------------
235 patient_tab = cpft_s1_tablename(S1Table.PATIENT)
236 cursor.execute(
237 f"""
238 SELECT
239 IDPatient, -- BIGINT; internal S1 reference number
240 Title,
241 FirstName,
242 GivenName2,
243 Surname,
244 DOB, -- DATETIME
245 DateDeath, -- DATETIME
246 DeathIndicator, -- INT
247 Gender, -- CHAR(1)
248 EmailAddress -- VARCHAR
249 -- Note also: TestPatient, BOOLEAN
250 FROM {patient_tab}
251 WHERE NHSNumber = %s -- CHAR comparison; VARCHAR(10)
252 """,
253 [str(lookup.nhs_number)],
254 )
255 rows = dictfetchall(cursor)
256 if not rows:
257 decisions.append(f"NHS number not found in {patient_tab} table.")
258 return
259 if len(rows) > 1:
260 decisions.append("Two patients found with that NHS number; aborting.")
261 return
262 row = rows[0]
263 s1_patient_id = row["IDPatient"]
264 secret_decisions.append(f"SystmOne patient ID: {s1_patient_id}")
265 lookup.pt_found = True
266 lookup.pt_local_id_description = "SystmOne patient ID"
267 lookup.pt_local_id_number = s1_patient_id
268 lookup.pt_title = row["Title"] or ""
269 lookup.pt_first_name = row["FirstName"] or ""
270 lookup.pt_last_name = row["Surname"] or ""
271 lookup.pt_dob = to_date(row["DOB"])
272 lookup.pt_dod = to_date(row["DateDeath"])
273 lookup.pt_dead = bool(lookup.pt_dod or row["DeathIndicator"])
274 lookup.pt_sex = _GENDER_S1_TO_CRATE.get(
275 row["Gender"], PatientLookup.UNKNOWNSEX
276 )
277 lookup.pt_email = row["EmailAddress"] or ""
279 # Deal with dodgy case
280 lookup.pt_title = lookup.pt_title.title()
281 lookup.pt_first_name = lookup.pt_first_name.title()
282 lookup.pt_last_name = lookup.pt_last_name.title()
284 # - There is also SRPatientContactDetails, but no e-mail address there;
285 # it's about phone numbers, I think.
287 # -------------------------------------------------------------------------
288 # 2. Address
289 # -------------------------------------------------------------------------
290 address_tab = cpft_s1_tablename(S1Table.ADDRESS_HISTORY)
291 cursor.execute(
292 f"""
293 SELECT
294 NameOfBuilding,
295 NumberOfBuilding, -- text
296 NameOfRoad,
297 NameOfLocality,
298 NameOfTown,
299 NameOfCounty,
300 FullPostCode
301 FROM {address_tab}
302 WHERE
303 IDPatient = %s
304 AND DateTo IS NULL -- still current
305 ORDER BY
306 DateEvent DESC -- most recent first
307 """,
308 [s1_patient_id],
309 )
310 row = dictfetchone(cursor)
311 if not row:
312 decisions.append(f"No address found in {address_tab} table.")
313 else:
314 lookup.pt_address_1 = row["NameOfBuilding"] or ""
315 lookup.pt_address_2 = " ".join(
316 filter(
317 None, [row["NumberOfBuilding"] or "", row["NameOfRoad"] or ""]
318 )
319 )
320 lookup.pt_address_3 = row["NameOfLocality"] or ""
321 lookup.pt_address_4 = row["NameOfTown"] or ""
322 lookup.pt_address_5 = row["NameOfCounty"] or ""
323 lookup.pt_address_6 = row["FullPostCode"] or ""
325 # -------------------------------------------------------------------------
326 # 3. GP
327 # -------------------------------------------------------------------------
328 # In the original SRE, this is SRGPPracticeHistory.
329 #
330 # _tmp = """
331 # SELECT *
332 # FROM information_schema.tables
333 # WHERE table_catalog = 'SystmOne'
334 # AND (
335 # table_name LIKE '%practi%'
336 # OR table_name LIKE '%gp%'
337 # )
338 # """
339 #
340 # ... it's S1_PatientGPPractice.
341 #
342 # In the original, SRGPPracticeHistory.IDPractice is a textual foreign key
343 # to SROrganisation.ID. However, in the CPFT copy, that's gone. Instead,
344 # there is S1_PatientGPPractice.Practice_Name, but there are no "%org%"
345 # tables. So we will get some, but limited, GP information.
347 gp_tab = cpft_s1_tablename(S1Table.GP_PRACTICE_HISTORY)
348 cursor.execute(
349 f"""
350 SELECT
351 IDPractice, -- FK to SROrganisation; text
352 Practice_Name
353 FROM {gp_tab}
354 WHERE
355 IDPatient = %s
356 AND DateTo IS NULL -- still current
357 ORDER BY
358 DateFrom DESC -- most recent first (unlikely >1 current!)
359 """,
360 [s1_patient_id],
361 )
362 row = dictfetchone(cursor)
363 if not row:
364 decisions.append(f"No GP found in {gp_tab} table.")
365 else:
366 lookup.gp_found = True
367 lookup.gp_address_1 = row["Practice_Name"] or ""
369 # -------------------------------------------------------------------------
370 # 4. CPFT clinician, active v. discharged
371 # -------------------------------------------------------------------------
372 # - PROBLEM: there appear to be no rows with staff e-mail addresses in the
373 # SRE -- nor in CPFT's copy.
374 # - So this will be lame.
375 # - We could in theory guess them (forename.surname@cpft.nhs.uk) but that
376 # is risky. However, team representatives will have proper e-mails
377 # recorded, in CRATE.
379 clinicians = [] # type: List[ClinicianInfoHolder]
381 # (a) Care coordinator?
382 care_co_tab = cpft_s1_tablename(S1Table.RESPONSIBLE_PARTY)
383 cursor.execute(
384 f"""
385 SELECT
386 IDProfileResponsibleParty, -- BIGINT NULL
387 Staff, -- firstname surname
388 DateStart,
389 DateEnd
390 FROM {care_co_tab}
391 WHERE
392 IDPatient = %s
393 AND Start_Date <= GETDATE()
394 """,
395 [s1_patient_id],
396 )
397 for row in dictfetchall(cursor):
398 care_co = _get_staff_details(
399 cursor=cursor,
400 clinician_type=ClinicianInfoHolder.CARE_COORDINATOR,
401 signatory_title=SignatoryTitles.CARE_COORDINATOR,
402 is_consultant=False, # We don't know. Assume not (for CTIMPs).
403 name=row["StaffName"],
404 profile_id=row["IDProfileResponsibleParty"],
405 start_date=to_date(row["DateStart"]),
406 end_date=to_date(row["DateEnd"]),
407 )
408 if care_co:
409 clinicians.append(care_co)
411 # (b) Active named consultant referral?
412 # - S1_Diagnosis, from SRClinicalCode
413 codes_tab = cpft_s1_tablename(S1Table.CLINICAL_CODE)
414 cursor.execute(
415 f"""
416 SELECT
417 IDProfileConsultant, -- BIGINT NULL
418 DateEpisodeStart,
419 DateEpisodeEnd
420 FROM {codes_tab}
421 WHERE
422 IDPatient = %s
423 """,
424 [s1_patient_id],
425 )
426 # ... NB also IDConsultantEvent, likely FK to
427 # SRHospitalConsultantEvent.RowIdentifier (in CPFT,
428 # S1_InpatientSpells_ConsultantEpisode), but we deal with that separately
429 # below. There is no staff name field here.
430 for row in dictfetchall(cursor):
431 _process_consultant(
432 clinicians=clinicians,
433 cursor=cursor,
434 profile_id=row["IDProfileConsultant"],
435 start_date=to_date(row["DateEpisodeStart"]),
436 end_date=to_date(row["DateEpisodeEnd"]),
437 )
438 # - S1_InpatientSpells_ConsultantEpisode
439 inpatient_tab = cpft_s1_tablename(S1Table.HOSPITAL_CONSULTANT_EVENT)
440 cursor.execute(
441 f"""
442 SELECT
443 IDProfileConsultant,
444 StaffName, -- is of the consultant
445 DateEpisodeStart,
446 DateEpisodeEnd
447 FROM {inpatient_tab}
448 WHERE
449 IDPatient = %s
450 """,
451 [s1_patient_id],
452 )
453 for row in dictfetchall(cursor):
454 _process_consultant(
455 clinicians=clinicians,
456 cursor=cursor,
457 name=row["StaffName"],
458 profile_id=row["IDProfileConsultant"],
459 start_date=to_date(row["DateEpisodeStart"]),
460 end_date=to_date(row["DateEpisodeEnd"]),
461 )
463 # (c) Active other named staff referral?
464 referral_staff_tab = cpft_s1_tablename(S1Table.REFERRAL_ALLOCATION)
465 # - In SRReferralAllocation, there is StaffName (VARCHAR) but we want more
466 # detail.
467 # - There is IDProfileStaffMember, in the original (FK to
468 # SRStaffMemberProfile.RowIdentifier). SRStaffMemberProfile doesn't
469 # contain names (but does contain e.g. roles, employment start/end
470 # dates).
471 # - There is also IDStaffMember, not in the original
472 # - Of 79566 rows in CPFT's S1_ReferralAllocation during testing,
473 # IDProfileStaffMember and IDStaffMember are always different!
474 cursor.execute(
475 f"""
476 SELECT
477 IDProfileStaffMember,
478 StaffName,
479 DateStart,
480 DateEnd
481 FROM {referral_staff_tab}
482 WHERE
483 IDPatient = %s
484 AND DateStart <= GETDATE()
485 AND DateDeleted IS NULL
486 """,
487 [s1_patient_id],
488 )
489 for row in dictfetchall(cursor):
490 hcp = _get_staff_details(
491 cursor=cursor,
492 clinician_type=ClinicianInfoHolder.HCP,
493 signatory_title=SignatoryTitles.CLINICIAN,
494 is_consultant=False, # We don't know. Assume not (for CTIMPs).
495 name=row["StaffName"],
496 profile_id=row["IDProfileStaffMember"],
497 start_date=to_date(row["DateStart"]),
498 end_date=to_date(row["DateEnd"]),
499 )
500 if hcp:
501 clinicians.append(hcp)
503 # (d) Active team referral?
504 referral_team_tab = cpft_s1_tablename(S1Table.REFERRAL_ALLOCATION)
505 cursor.execute(
506 f"""
507 SELECT
508 TeamName,
509 DateStart,
510 DateEnd
511 FROM {referral_team_tab}
512 WHERE
513 IDPatient = %s
514 AND DateStart <= GETDATE()
515 AND DateDeleted IS NULL
516 """,
517 [s1_patient_id],
518 )
519 for row in dictfetchall(cursor):
520 team_info = get_team_details(
521 team_name=row["TeamName"] or "",
522 start_date=to_date(row["DateStart"]),
523 end_date=to_date(row["DateEnd"]),
524 decisions=decisions,
525 )
526 clinicians.append(team_info)
527 # We append it even if we can't find a representative, because it still
528 # carries information about whether the patient is discharged or not.
530 # Now pick one:
531 pick_best_clinician(lookup, clinicians, decisions)
534# =============================================================================
535# Look up choices about research consent
536# =============================================================================
539class ResearchSNOMED:
540 """
541 SNOMED codes used for the CPFT Research Database in SystmOne.
543 Verify at https://termbrowser.nhs.uk/.
544 """
546 RED = 752581000000107
547 # 752581000000107 | Declined consent for researcher to access clinical
548 # record (finding) |
550 YELLOW = 871801000000106
551 # 871801000000106 | Possibly eligible for participation in research study
552 # (finding) |
554 GREEN = 873791000000106
555 # 873791000000106 | Consent given to review medical record in research
556 # study (finding) |
558 EMAIL = 705025004
559 # 705025004 | Consent given for communication by email (finding) |
561 NO_EMAIL = 835231000000104
562 # 835231000000104 | Declined consent for communication by email (finding) |
564 OPT_OUT = 1091881000000109
565 # 1091881000000109 | Invitation to participate in research study declined
566 # (situation) |
569def get_latest_consent_mode_from_cpft_systmone(
570 nhs_number: int,
571 decisions: List[str],
572) -> Optional[ConsentMode]:
573 """
574 Returns the latest CPFT consent mode for a patient, from a CPFT SystmOne
575 database.
577 Args:
578 nhs_number: NHS number
579 decisions: list of human-readable decisions; will be modified
581 Returns:
582 a :class:`crate_anon.crateweb.consent.models.ConsentMode`, or ``None``
583 """
584 cursor = connections[ClinicalDatabaseType.CPFT_SYSTMONE].cursor()
585 patient_tab = cpft_s1_tablename(S1Table.PATIENT)
586 prefs_tab = cpft_s1_tablename(CPFTTable.CPFTRD_PREFS)
587 email_tab = cpft_s1_tablename(CPFTTable.CPFTRD_EMAIL)
588 opt_out_tab = cpft_s1_tablename(CPFTTable.CPFTRD_OPT_OUT)
590 # (a) Establish SystmOne patient ID (required)/
591 cursor.execute(
592 f"""
593 SELECT IDPatient
594 FROM {patient_tab}
595 WHERE
596 NHSNumber = %s -- CHAR comparison; VARCHAR(10)
597 """,
598 [str(nhs_number)],
599 )
600 row = dictfetchone(cursor)
601 if not row:
602 decisions.append("Could not find patient by NHS number")
603 return None
604 patient_id = row["IDPatient"]
606 # (b) Traffic-light preferences (required).
607 cursor.execute(
608 f"""
609 SELECT TOP 1 SNOMEDCode, DateEvent
610 FROM {prefs_tab}
611 WHERE IDPatient = %s
612 ORDER BY DateEvent DESC -- most recent (for this patient) first
613 """,
614 [patient_id],
615 )
616 # Note also:
617 # - ResearchOption -- a textual version, e.g. "Red - Declined consent...";
618 # - EmailCheck: VARCHAR(100) -- e.g. "Email - Consent given...", but see
619 # below;
620 # - OptOut: textual "Declined..." or NULL, but see below;
621 # - NationalDataOptOut (but that's for other purposes).
622 # Not shown:
623 # - age; capacity decisions; etc.
624 row = dictfetchone(cursor)
625 if not row:
626 decisions.append("No traffic-light decision found for patient")
627 return None
628 # Opt-outs will still be respected (via a separate process).
629 traffic_snomed = row["SNOMEDCode"]
630 traffic_light_when = to_date(row["DateEvent"])
631 if traffic_snomed == ResearchSNOMED.RED:
632 decisions.append("Patient has chosen RED")
633 traffic_light = ConsentMode.RED
634 elif traffic_snomed == ResearchSNOMED.YELLOW:
635 decisions.append("Patient has chosen YELLOW")
636 traffic_light = ConsentMode.YELLOW
637 elif traffic_snomed == ResearchSNOMED.GREEN:
638 decisions.append("Patient has chosen GREEN")
639 traffic_light = ConsentMode.GREEN
640 else:
641 decisions.append("Unknown SNOMED code for traffic-light decision")
642 return None
644 # (c) Email preferences (optional).
645 cursor.execute(
646 f"""
647 SELECT TOP 1 SNOMEDCode
648 FROM {email_tab}
649 WHERE IDPatient = %s
650 ORDER BY DateEvent DESC -- most recent (for this patient) first
651 """,
652 [patient_id],
653 )
654 row = dictfetchone(cursor)
655 prefers_email = False
656 if row:
657 email_snomed = row["SNOMEDCode"]
658 if email_snomed == ResearchSNOMED.EMAIL:
659 decisions.append("Patient prefers e-mail")
660 prefers_email = True
661 elif email_snomed == ResearchSNOMED.NO_EMAIL:
662 decisions.append("Patient prefers not to use e-mail")
663 else:
664 decisions.append("Unknown SNOMED code regarding e-mail")
665 else:
666 decisions.append("Could not find e-mail decision")
668 # (d) Opting out (excluding entirely)?
669 cursor.execute(
670 f"""
671 SELECT COUNT(*) AS n_opt_outs
672 FROM {opt_out_tab}
673 WHERE
674 IDPatient = %s
675 AND SNOMEDCode = %s -- likely a redundant check
676 """,
677 [patient_id, ResearchSNOMED.OPT_OUT],
678 )
679 row = dictfetchone(cursor)
680 n_opt_outs = row["n_opt_outs"]
681 if n_opt_outs > 0:
682 decisions.append("Opt-out found for patient")
683 exclude_entirely = True
684 else:
685 decisions.append("No opt-out found for patient")
686 exclude_entirely = False
688 # Compile results.
689 return ConsentMode(
690 nhs_number=nhs_number,
691 created_at=traffic_light_when, # source, not "now"; see lookup_rio.py
692 exclude_entirely=exclude_entirely,
693 consent_mode=traffic_light,
694 prefers_email=prefers_email,
695 # SystmOne doesn't provide the following information, but relies (as
696 # CPFT policy) on valid decisions being attested to by the clinician
697 # entering the information. The following is valid, and meets the
698 # Decision.decision_valid() test.
699 decision_signed_by_patient=False,
700 decision_otherwise_directly_authorized_by_patient=True,
701 decision_under16_signed_by_parent=False,
702 decision_under16_signed_by_clinician=False,
703 decision_lack_capacity_signed_by_representative=False,
704 decision_lack_capacity_signed_by_clinician=False,
705 )
708def gen_opt_out_pids_mpids_cpft_systmone() -> (
709 Generator[Tuple[str, str], None, None]
710):
711 """
712 Generates PID/MPID pairs from all patients opting out entirely from the
713 CPFT Research Database, from a CPFT SystmOne database.
715 Note: this is the CPFT Research Database opt-out, not the NHS National Data
716 Opt-Out. The latter applies to NHS Act s251 work, and is in
717 SRNDOptOutPreference [marked as "future" in the SRE details?] or, in CPFT's
718 copy, S1_Patient.NationalDataOptOut.
720 Yields:
721 ``s1_patient_id, nhs_number`` for each patient opting out, in string
722 format
724 PRESUMPTION: that if some "un-opts out", their row here is deleted. There
725 is a DateEvent column, but no codes for reversing the decision.
727 .. code-block:: sql
729 SELECT DISTINCT SNOMEDCode, SNOMEDText, CTV3Code, CTV3Text
730 FROM S1_ClinicalOutcome_ConsentResearch_OptOutCheck
731 """
732 cursor = connections[ClinicalDatabaseType.CPFT_SYSTMONE].cursor()
733 opt_out_tab = cpft_s1_tablename(CPFTTable.CPFTRD_OPT_OUT)
734 patient_tab = cpft_s1_tablename(S1Table.PATIENT)
735 cursor.execute(
736 f"""
737 SELECT
738 o.IDPatient, -- BIGINT, NULL (though probably not in the original)
739 p.NHSNumber
740 FROM {opt_out_tab} AS o
741 INNER JOIN {patient_tab} AS p
742 ON p.IDPatient = o.IDPatient
743 WHERE
744 o.IDPatient IS NOT NULL -- unlikely in practice!
745 AND o.SNOMEDCode = %s -- likely a redundant check
746 """,
747 [ResearchSNOMED.OPT_OUT],
748 )
749 for row in genrows(cursor):
750 pid = row[0] # SystmOne internal ID number, as text
751 mpid = row[1] # NHS number, as text
752 yield pid, mpid