Coverage for cc_modules/cc

1# noinspection HttpUrlsUsage

2"""

3camcops_server/cc_modules/cc_fhir.py

5===============================================================================

8 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).

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**Implements communication with a FHIR server.**

29Fast Healthcare Interoperability Resources

31https://www.hl7.org/fhir/

33Our implementation exports:

35- patients as FHIR Patient resources;

36- task concepts as FHIR Questionnaire resources;

37- task instances as FHIR QuestionnaireResponse resources.

39Currently PHQ9 and APEQPT (anonymous) are supported. Each task and patient (if

40appropriate is sent to the FHIR server in a single "transaction" Bundle).

41The resources are given a unique identifier based on the URL of the CamCOPS

42server.

44We use the Python client https://github.com/smart-on-fhir/client-py/.

45This only supports one version of the FHIR specification (currently 4.0.1).

48*Testing: HAPI FHIR server locally*

50To test with a HAPI FHIR server locally, which was installed from instructions

51at https://github.com/hapifhir/hapi-fhir-jpaserver-starter (Docker). Most

52simply:

54.. code-block:: bash

56 docker run -p 8080:8080 hapiproject/hapi:latest

58with the following entry in the CamCOPS export recipient configuration:

60.. code-block:: ini

62 FHIR_API_URL = http://localhost:8080/fhir

64To inspect it while it's running (apart from via its log):

66- Browse to (by default) http://localhost:8080/

68 - then e.g. Patient --> Search, which is a pretty version of

69 http://localhost:8080/fhir/Patient?_pretty=true;

71 - Questionnaire --> Search, which is a pretty version of

72 http://localhost:8080/fhir/Questionnaire?_pretty=true.

74- Can also browse to (by default) http://localhost:8080/fhir/metadata

77*Testing: Other*

79There are also public sandboxes at:

81- http://hapi.fhir.org/baseR4

82- https://r4.smarthealthit.org (errors when exporting questionnaire responses)

85*Intermittent problem with If-None-Exist*

87This problem occurs intermittently:

89- "Failed to CREATE resource with match URL ... because this search matched 2

90 resources" -- an OperationOutcome error.

92 At https://groups.google.com/g/hapi-fhir/c/8OolMOpf8SU, it says (for an error

93 with 40 resources) "You can only do a conditional create if there are 0..1

94 existing resources on the server that match the criteria, and in this case

95 there are 40." But I think that is an error in the explanation.

97 Proper documentation for ``ifNoneExist`` (Python client) or ``If-None-Exist``

98 (FHIR itself) is at https://www.hl7.org/fhir/http.html#ccreate.

100 I suspect that "0..1" comment relates to "cardinality"

101 (https://www.hl7.org/fhir/bundle.html#resource), which is how many times the

102 attribute can appear in a resource type

103 (https://www.hl7.org/fhir/conformance-rules.html#cardinality); that is, this

104 statement is optional. It would clearly be silly if it meant "create if no

105 more than 1 exist"!

106

107 However, the "Failed to CREATE" problem seemed to go away. It does work fine,

108 and you get status messages of "200 OK" rather than "201 Created" if you try

109 to insert the same information again (``SELECT * FROM

110 _exported_task_fhir_entry;``).

111

112- This is a concurrency problem (they dispute "bug") in the HAPI FHIR

113 implementation. See our bug report at

114 https://github.com/hapifhir/hapi-fhir/issues/3141.

115

116- The suggested fix is a "unique combo search index parameter", as per

117 https://smilecdr.com/docs/fhir_repository/custom_search_parameters.html#uniqueness.

118

119- However, that seems implementation-specific (e.g. HAPI FHIR, SmileCDR). A

120 specific value of ``http://hapifhir.io/fhir/StructureDefinition/sp-unique``

121 must be used. Specimen code is

122 https://hapifhir.io/hapi-fhir/apidocs/hapi-fhir-base/src-html/ca/uhn/fhir/util/HapiExtensions.html.

123

124- Instead, we could force a FHIR export for a given recipient to occur in

125 serial (particularly as other FHIR implementations may have this bug).

126

127 Celery doesn't allow you to send multiple jobs and enforce that they all

128 happen via the same worker

129 (https://docs.celeryproject.org/en/stable/userguide/calling.html). However,

130 our exports already start (mostly!) as "one recipient, one job", via

131 :func:`camcops_server.cc_modules.celery.export_to_recipient_backend` (see

132 :func:`camcops_server.cc_modules.celery.get_celery_settings_dict`).

133

134 The tricky bit is that push exports require back-end single-task jobs, so

135 they are hard to de-parallelize.

136

137 So we use a carefully sequenced file lock; see

138 :func:`camcops_server.cc_modules.cc_export.export_task`.

139

140"""

141

142

143# =============================================================================

144# Imports

145# =============================================================================

146

147from enum import Enum

148import json

149import logging

150from typing import Any, Dict, List, TYPE_CHECKING

151

152from cardinal_pythonlib.datetimefunc import format_datetime

153from cardinal_pythonlib.httpconst import HttpMethod

154from fhirclient.client import FHIRClient

155from fhirclient.models.bundle import Bundle, BundleEntry, BundleEntryRequest

156from fhirclient.models.codeableconcept import CodeableConcept

157from fhirclient.models.coding import Coding

158from fhirclient.models.fhirdate import FHIRDate

159from fhirclient.models.identifier import Identifier

160from fhirclient.models.observation import ObservationComponent

161from fhirclient.models.questionnaire import (

162 QuestionnaireItem,

163 QuestionnaireItemAnswerOption,

164)

165from fhirclient.models.quantity import Quantity

166from fhirclient.models.questionnaireresponse import (

167 QuestionnaireResponseItem,

168 QuestionnaireResponseItemAnswer,

169)

170from requests.exceptions import HTTPError

171

172from camcops_server.cc_modules.cc_constants import (

173 DateFormat,

174 FHIRConst as Fc,

175 JSON_INDENT,

176)

177from camcops_server.cc_modules.cc_exception import FhirExportException

178from camcops_server.cc_modules.cc_pyramid import Routes

179from camcops_server.cc_modules.cc_snomed import SnomedExpression, SnomedLookup

180

181if TYPE_CHECKING:

182 from camcops_server.cc_modules.cc_exportmodels import ExportedTaskFhir

183 from camcops_server.cc_modules.cc_request import CamcopsRequest

184

185log = logging.getLogger(__name__)

186

187

188# =============================================================================

189# Debugging options

190# =============================================================================

191

192DEBUG_FHIR_TX = False # needs workers to be launched with "--verbose" option

193

194if any([DEBUG_FHIR_TX]):

195 log.warning("Debugging options enabled!")

196

197

198# =============================================================================

199# Development thoughts

200# =============================================================================

201

202_ = r"""

203

204Dive into the internals of the HAPI FHIR server

205===============================================

206

207.. code-block:: bash

208

209 docker container ls | grep hapi # find its container ID

210 docker exec -it <CONTAINER_NAME_OR_ID> bash

211

212 # Find files modified in the last 10 minutes:

213 find / -mmin -10 -type f -not -path "/proc/*" -not -path "/sys/*" -exec ls -l {} \;

214 # ... which reveals /usr/local/tomcat/target/database/h2.mv.db

215 # and /usr/local/tomcat/logs/localhost_access_log*

216

217 # Now, from https://h2database.com/html/tutorial.html#command_line_tools,

218 find / -name "h2*.jar"

219 # ... /usr/local/tomcat/webapps/ROOT/WEB-INF/lib/h2-1.4.200.jar

220

221 java -cp /usr/local/tomcat/webapps/ROOT/WEB-INF/lib/h2*.jar org.h2.tools.Shell

222 # - URL = jdbc:h2:/usr/local/tomcat/target/database/h2

223 # ... it will append ".mv.db"

224 # - Accept other defaults.

225 # - Then from the "sql>" prompt, try e.g. SHOW TABLES;

226

227However, it won't connect with the server open. (And you can't stop the Docker

228FHIR server and repeat the connection using ``docker run -it <IMAGE_ID> bash``

229rather than ``docker exec``, because then the data will disappear as Docker

230returns to its starting image.) But you can copy the database and open the

231copy, e.g. with

232

233.. code-block:: bash

234

235 cd /usr/local/tomcat/target/database

236 cp h2.mv.db h2_copy.mv.db

237 java -cp /usr/local/tomcat/webapps/ROOT/WEB-INF/lib/h2*.jar org.h2.tools.Shell

238 # URL = jdbc:h2:/usr/local/tomcat/target/database/h2_copy

239

240but then that needs a username/password. Better is to create

241``application.yaml`` in a host machine directory, like this:

242

243.. code-block:: bash

244

245 # From MySQL:

246 # CREATE DATABASE hapi_test_db;

247 # CREATE USER 'hapi_test_user'@'localhost' IDENTIFIED BY 'hapi_test_password';

248 # GRANT ALL PRIVILEGES ON hapi_test_db.* TO 'hapi_test_user'@'localhost';

249

250 mkdir ~/hapi_test

251 cd ~/hapi_test

252 git clone https://github.com/hapifhir/hapi-fhir-jpaserver-starter

253 cd hapi-fhir-jpaserver-starter

254 nano src/main/resources/application.yaml

255

256... no, better is to use the web interface!

257

258

259Wipe FHIR exports

260=================

261

262.. code-block:: sql

263

264 -- Delete all records of tasks exported via FHIR:

265 DELETE FROM _exported_task_fhir_entry;

266 DELETE FROM _exported_task_fhir;

267 DELETE FROM _exported_tasks WHERE recipient_id IN (

268 SELECT id FROM _export_recipients WHERE transmission_method = 'fhir'

269 );

270

271 -- Delete ALL export information

272 DELETE FROM _exported_task_fhir_entry;

273 DELETE FROM _exported_task_fhir;

274 DELETE FROM _exported_task_email;

275 DELETE FROM _exported_task_filegroup;

276 DELETE FROM _exported_task_hl7msg;

277 DELETE FROM _exported_task_redcap;

278 DELETE FROM _exported_tasks;

279 DELETE FROM _export_recipients;

280

281What's been sent?

282

283.. code-block:: sql

284

285 -- Tasks exported via FHIR:

286 SELECT * FROM _exported_tasks WHERE recipient_id IN (

287 SELECT id FROM _export_recipients WHERE transmission_method = 'fhir'

288 );

289

290 -- Entries for all BMI tasks:

291 SELECT * FROM _exported_task_fhir_entry WHERE exported_task_fhir_id IN (

292 SELECT _exported_task_fhir.id FROM _exported_task_fhir

293 INNER JOIN _exported_tasks

294 ON _exported_task_fhir.exported_task_id = _exported_tasks.id

295 INNER JOIN _export_recipients

296 ON _exported_tasks.recipient_id = _export_recipients.id

297 WHERE _export_recipients.transmission_method = 'fhir'

298 AND _exported_tasks.basetable = 'bmi'

299 );

300

301

302Inspecting fhirclient

303=====================

304

305Each class has entries like this:

306

307.. code-block:: python

308

309 def elementProperties(self):

310 js = super(DocumentReference, self).elementProperties()

311 js.extend([

312 ("authenticator", "authenticator", fhirreference.FHIRReference, False, None, False),

313 ("author", "author", fhirreference.FHIRReference, True, None, False),

314 # ...

315 ])

316 return js

317

318The fields are: ``name, jsname, typ, is_list, of_many, not_optional``.

319They are validated in FHIRAbstractBase.update_with_json().

320

321""" # noqa: E501

322

323

324# =============================================================================

325# Export tasks via FHIR

326# =============================================================================

327

328

329class FhirTaskExporter(object):

330 """

331 Class that knows how to export a single task to FHIR.

332 """

333

334 def __init__(

335 self, request: "CamcopsRequest", exported_task_fhir: "ExportedTaskFhir"

336 ) -> None:

337 self.request = request

338 self.exported_task = exported_task_fhir.exported_task

339 self.exported_task_fhir = exported_task_fhir

340

341 self.recipient = self.exported_task.recipient

342 self.task = self.exported_task.task

343

344 # TODO: In theory these settings should handle authentication

345 # for any server that is SMART-compliant but we've not tested this.

346 # https://sep.com/blog/smart-on-fhir-what-is-smart-what-is-fhir/

347 settings = {

348 Fc.API_BASE: self.recipient.fhir_api_url,

349 Fc.APP_ID: self.recipient.fhir_app_id,

350 Fc.APP_SECRET: self.recipient.fhir_app_secret,

351 Fc.LAUNCH_TOKEN: self.recipient.fhir_launch_token,

352 }

353

354 try:

355 self.client = FHIRClient(settings=settings)

356 except Exception as e:

357 raise FhirExportException(f"Error creating FHIRClient: {e}")

358

359 def export_task(self) -> None:

360 """

361 Export a single task to the server, with associated patient information

362 if the task has an associated patient.

363 """

364

365 # TODO: Check FHIR server's capability statement

366 # https://www.hl7.org/fhir/capabilitystatement.html

367 #

368 # statement = self.client.server.capabilityStatement

369 # The client doesn't support looking for a particular capability

370 # We could check for:

371 # fhirVersion (the client does not support multiple versions)

372 # conditional create

373 # supported resource types (statement.rest[0].resource[])

374

375 bundle = self.task.get_fhir_bundle(

376 self.request, self.exported_task.recipient

377 ) # may raise FhirExportException

378

379 try:

380 # Attempt to create the receiver on the server, via POST:

381 if DEBUG_FHIR_TX:

382 bundle_str = json.dumps(bundle.as_json(), indent=JSON_INDENT)

383 log.debug(f"FHIR bundle outbound to server:\n{bundle_str}")

384 response = bundle.create(self.client.server)

385 if response is None:

386 # Not sure this will ever happen.

387 # fhirabstractresource.py create() says it returns

388 # "None or the response JSON on success" but an exception will

389 # already have been raised if there was a failure

390 raise FhirExportException(

391 "The FHIR server unexpectedly returned an OK, empty "

392 "response"

393 )

394

395 self.parse_response(response)

396

397 except HTTPError as e:

398 raise FhirExportException(

399 f"The FHIR server returned an error: {e.response.text}"

400 )

401

402 except Exception as e:

403 # Unfortunate that fhirclient doesn't give us anything more

404 # specific

405 raise FhirExportException(f"Error from fhirclient: {e}")

406

407 def parse_response(self, response: Dict) -> None:

408 """

409 Parse the response from the FHIR server to which we have sent our

410 task. The response looks something like this:

411

412 .. code-block:: json

413

414 {

415 "resourceType": "Bundle",

416 "id": "cae48957-e7e6-4649-97f8-0a882076ad0a",

417 "type": "transaction-response",

418 "link": [

419 {

420 "relation": "self",

421 "url": "http://localhost:8080/fhir"

422 }

423 ],

424 "entry": [

425 {

426 "response": {

427 "status": "200 OK",

428 "location": "Patient/1/_history/1",

429 "etag": "1"

430 }

431 },

432 {

433 "response": {

434 "status": "200 OK",

435 "location": "Questionnaire/26/_history/1",

436 "etag": "1"

437 }

438 },

439 {

440 "response": {

441 "status": "201 Created",

442 "location": "QuestionnaireResponse/42/_history/1",

443 "etag": "1",

444 "lastModified": "2021-05-24T09:30:11.098+00:00"

445 }

446 }

447 ]

448 }

449

450 The server's reply contains a Bundle

451 (https://www.hl7.org/fhir/bundle.html), which is a container for

452 resources. Here, the bundle contains entry objects

453 (https://www.hl7.org/fhir/bundle-definitions.html#Bundle.entry).

454

455 """

456 bundle = Bundle(jsondict=response)

457

458 if bundle.entry is not None:

459 self._save_exported_entries(bundle)

460

461 def _save_exported_entries(self, bundle: Bundle) -> None:

462 """

463 Record the server's reply components in strucured format.

464 """

465 from camcops_server.cc_modules.cc_exportmodels import (

466 ExportedTaskFhirEntry,

467 ) # delayed import

468

469 for entry in bundle.entry:

470 saved_entry = ExportedTaskFhirEntry()

471 saved_entry.exported_task_fhir_id = self.exported_task_fhir.id

472 saved_entry.status = entry.response.status

473 saved_entry.location = entry.response.location

474 saved_entry.etag = entry.response.etag

475 if entry.response.lastModified is not None:

476 # ... of type :class:`fhirclient.models.fhirdate.FHIRDate`

477 saved_entry.last_modified = entry.response.lastModified.date

478

479 self.request.dbsession.add(saved_entry)

480

481

482# =============================================================================

483# Helper functions for building FHIR component objects

484# =============================================================================

485

486

487def fhir_pk_identifier(

488 req: "CamcopsRequest", tablename: str, pk: int, value_within_task: str

489) -> Identifier:

490 """

491 Creates a "fallback" identifier -- this is poor, but allows unique

492 identification of anything (such as a patient with no proper ID numbers)

493 based on its CamCOPS table name and server PK.

494 """

495 return Identifier(

496 jsondict={

497 Fc.SYSTEM: req.route_url(

498 Routes.FHIR_TABLENAME_PK_ID, table_name=tablename, server_pk=pk

499 ),

500 Fc.VALUE: value_within_task,

501 }

502 )

503

504

505def fhir_system_value(system: str, value: str) -> str:

506 """

507 How FHIR expresses system/value pairs.

508 """

509 return f"{system}|{value}"

510

511

512def fhir_sysval_from_id(identifier: Identifier) -> str:

513 """

514 How FHIR expresses system/value pairs.

515 """

516 return f"{identifier.system}|{identifier.value}"

517

518

519def fhir_reference_from_identifier(identifier: Identifier) -> str:

520 """

521 Returns a reference to a specific FHIR identifier.

522 """

523 return f"{Fc.IDENTIFIER}={fhir_sysval_from_id(identifier)}"

524

525

526def fhir_observation_component_from_snomed(

527 req: "CamcopsRequest", expr: SnomedExpression

528) -> Dict:

529 """

530 Returns a FHIR ObservationComponent (as a dict in JSON format) for a SNOMED

531 CT expression.

532 """

533 observable_entity = req.snomed(SnomedLookup.OBSERVABLE_ENTITY)

534 expr_longform = expr.as_string(longform=True)

535 # For SNOMED, we are providing an observation where the "value" is a code

536 # -- thus, we use "valueCodeableConcept" as the specific value (the generic

537 # being "value<something>" or what FHIR calls "value[x]"). But there also

538 # needs to be a coding system, specified via "code".

539 return ObservationComponent(

540 jsondict={

541 # code = "the type of thing reported here"

542 # Per https://www.hl7.org/fhir/observation.html#code-interop, we

543 # use SNOMED 363787002 = Observable entity.

544 Fc.CODE: CodeableConcept(

545 jsondict={

546 Fc.CODING: [

547 Coding(

548 jsondict={

549 Fc.SYSTEM: Fc.CODE_SYSTEM_SNOMED_CT,

550 Fc.CODE: str(observable_entity.identifier),

551 Fc.DISPLAY: observable_entity.as_string(

552 longform=True

553 ),

554 Fc.USER_SELECTED: False,

555 }

556 ).as_json()

557 ],

558 Fc.TEXT: observable_entity.term,

559 }

560 ).as_json(),

561 # value = "the value of the thing"; the actual SNOMED code of

562 # interest:

563 Fc.VALUE_CODEABLE_CONCEPT: CodeableConcept(

564 jsondict={

565 Fc.CODING: [

566 Coding(

567 jsondict={

568 # http://www.hl7.org/fhir/snomedct.html

569 Fc.SYSTEM: Fc.CODE_SYSTEM_SNOMED_CT,

570 Fc.CODE: expr.as_string(longform=False),

571 Fc.DISPLAY: expr_longform,

572 Fc.USER_SELECTED: False,

573 # ... means "did the user choose it

574 # themselves?"

575 # version: not used

576 }

577 ).as_json()

578 ],

579 Fc.TEXT: expr_longform,

580 }

581 ).as_json(),

582 }

583 ).as_json()

584

585

586def make_fhir_bundle_entry(

587 resource_type_url: str,

588 identifier: Identifier,

589 resource: Dict,

590 identifier_is_list: bool = True,

591) -> Dict:

592 """

593 Builds a FHIR BundleEntry, as a JSON dict.

594

595 This also takes care of the identifier, by ensuring (a) that the resource

596 is labelled with the identifier, and (b) that the BundleEntryRequest has

597 an ifNoneExist condition referring to that identifier.

598 """

599 if Fc.IDENTIFIER in resource:

600 log.warning(

601 f"Duplication: {Fc.IDENTIFIER!r} specified in resource "

602 f"but would be auto-added by make_fhir_bundle_entry()"

603 )

604 if identifier_is_list:

605 # Some, like Observation, Patient, and Questionnaire, need lists here.

606 resource[Fc.IDENTIFIER] = [identifier.as_json()]

607 else:

608 # Others, like QuestionnaireResponse, don't.

609 resource[Fc.IDENTIFIER] = identifier.as_json()

610 bundle_request = BundleEntryRequest(

611 jsondict={

612 Fc.METHOD: HttpMethod.POST,

613 Fc.URL: resource_type_url,

614 Fc.IF_NONE_EXIST: fhir_reference_from_identifier(identifier),

615 # "If this resource doesn't exist, as determined by this

616 # identifier, then create it:"

617 # https://www.hl7.org/fhir/http.html#ccreate

618 }

619 )

620 return BundleEntry(

621 jsondict={Fc.REQUEST: bundle_request.as_json(), Fc.RESOURCE: resource}

622 ).as_json()

623

624

625# =============================================================================

626# Helper classes for building FHIR component objects

627# =============================================================================

628

629

630class FHIRQuestionType(Enum):

631 """

632 An enum for value type keys of QuestionnaireResponseItemAnswer.

633 """

634

635 ATTACHMENT = Fc.QITEM_TYPE_ATTACHMENT

636 BOOLEAN = Fc.QITEM_TYPE_BOOLEAN

637 CHOICE = Fc.QITEM_TYPE_CHOICE

638 DATE = Fc.QITEM_TYPE_DATE

639 DATETIME = Fc.QITEM_TYPE_DATETIME

640 DECIMAL = Fc.QITEM_TYPE_DECIMAL

641 DISPLAY = Fc.QITEM_TYPE_DISPLAY

642 GROUP = Fc.QITEM_TYPE_GROUP

643 INTEGER = Fc.QITEM_TYPE_INTEGER

644 OPEN_CHOICE = Fc.QITEM_TYPE_OPEN_CHOICE

645 QUANTITY = Fc.QITEM_TYPE_QUANTITY

646 QUESTION = Fc.QITEM_TYPE_QUESTION

647 REFERENCE = Fc.QITEM_TYPE_REFERENCE

648 STRING = Fc.QITEM_TYPE_STRING

649 TIME = Fc.QITEM_TYPE_TIME

650 URL = Fc.QITEM_TYPE_URL

651

652

653class FHIRAnswerType(Enum):

654 """

655 An enum for value type keys of QuestionnaireResponseItemAnswer.

656 """

657

658 ATTACHMENT = Fc.VALUE_ATTACHMENT

659 BOOLEAN = Fc.VALUE_BOOLEAN

660 CODING = Fc.VALUE_CODING

661 DATE = Fc.VALUE_DATE

662 DATETIME = Fc.VALUE_DATETIME

663 DECIMAL = Fc.VALUE_DECIMAL

664 INTEGER = Fc.VALUE_INTEGER

665 QUANTITY = Fc.VALUE_QUANTITY # e.g. real number

666 REFERENCE = Fc.VALUE_REFERENCE

667 STRING = Fc.VALUE_STRING

668 TIME = Fc.VALUE_TIME

669 URI = Fc.VALUE_URI

670

671

672class FHIRAnsweredQuestion:

673 """

674 Represents a question in a questionnaire-based task. That includes both the

675 abstract aspects:

676

677 - What kind of question is it (e.g. multiple-choice, real-value answer,

678 text)? That can go into some detail, e.g. possible responses for a

679 multiple-choice question. (Thus, the FHIR Questionnaire.)

680

681 and the concrete aspects:

682

683 - what is the response/answer for a specific task instance?

684 (Thus, the FHIR QuestionnaireResponse.)

685

686 Used for autodiscovery.

687 """

688

689 def __init__(

690 self,

691 qname: str,

692 qtext: str,

693 qtype: FHIRQuestionType,

694 answer_type: FHIRAnswerType,

695 answer: Any,

696 answer_options: Dict[Any, str] = None,

697 ) -> None:

698 """

699 Args:

700 qname:

701 Name (task attribute name) of the question, e.g. "q1".

702 qtext:

703 Question text (e.g. "How was your day?").

704 qtype:

705 Question type, e.g. multiple-choice.

706 answer_type:

707 Answer type, e.g. integer.

708 answer:

709 Actual answer.

710 answer_options:

711 For multiple-choice questions (MCQs), a dictionary mapping

712 answer codes to human-legible display text.

713 """

714 self.qname = qname

715 self.qtext = qtext

716 self.qtype = qtype

717 self.answer = answer

718 self.answer_type = answer_type

719 self.answer_options = answer_options or {} # type: Dict[Any, str]

720

721 # Checks

722 if self.is_mcq:

723 assert self.answer_options, (

724 f"Multiple choice item {self.qname!r} needs mcq_qa parameter, "

725 f"currently {answer_options!r}"

726 )

727

728 def __str__(self) -> str:

729 if self.is_mcq:

730 options = " / ".join(

731 f"{code} = {display}"

732 for code, display in self.answer_options.items()

733 )

734 else:

735 options = "N/A"

736 return (

737 f"{self.qname} "

738 f"// QUESTION: {self.qtext} "

739 f"// OPTIONS: {options} "

740 f"// ANSWER: {self.answer!r}, of type {self.answer_type.value}"

741 )

742

743 @property

744 def is_mcq(self) -> bool:

745 """

746 Is this a multiple-choice question?

747 """

748 return self.qtype in (

749 FHIRQuestionType.CHOICE,

750 FHIRQuestionType.OPEN_CHOICE,

751 )

752

753 # -------------------------------------------------------------------------

754 # Abstract (class)

755 # -------------------------------------------------------------------------

756

757 def questionnaire_item(self) -> Dict:

758 """

759 Returns a JSON/dict representation of a FHIR QuestionnaireItem.

760 """

761 qtype = self.qtype

762 # Basics

763 qitem_dict = {

764 Fc.LINK_ID: self.qname,

765 Fc.TEXT: self.qtext,

766 Fc.TYPE: qtype.value,

767 }

768

769 # Extras for multiple-choice questions: what are the possible answers?

770 if self.is_mcq:

771 # Add permitted answers.

772 options = [] # type: List[Dict]

773 # We asserted mcq_qa earlier.

774 for code, display in self.answer_options.items():

775 options.append(

776 QuestionnaireItemAnswerOption(

777 jsondict={

778 Fc.VALUE_CODING: {

779 Fc.CODE: str(code),

780 Fc.DISPLAY: display,

781 }

782 }

783 ).as_json()

784 )

785 qitem_dict[Fc.ANSWER_OPTION] = options # type: ignore[assignment]

786

787 return QuestionnaireItem(jsondict=qitem_dict).as_json()

788

789 # -------------------------------------------------------------------------

790 # Concrete (instance)

791 # -------------------------------------------------------------------------

792

793 def _qr_item_answer(self) -> QuestionnaireResponseItemAnswer:

794 """

795 Returns a QuestionnaireResponseItemAnswer.

796 """

797 # Look things up

798 raw_answer = self.answer

799 answer_type = self.answer_type

800

801 # Convert the value

802 if raw_answer is None:

803 # Deal with null values first, otherwise we will get

804 # mis-conversion, e.g. str(None) == "None", bool(None) == False.

805 fhir_answer = None

806 elif answer_type == FHIRAnswerType.BOOLEAN:

807 fhir_answer = bool(raw_answer)

808 elif answer_type == FHIRAnswerType.DATE:

809 fhir_answer = FHIRDate(

810 format_datetime(raw_answer, DateFormat.FHIR_DATE)

811 ).as_json()

812 elif answer_type == FHIRAnswerType.DATETIME:

813 fhir_answer = FHIRDate(raw_answer.isoformat()).as_json()

814 elif answer_type == FHIRAnswerType.DECIMAL:

815 fhir_answer = float(raw_answer) # type: ignore[assignment]

816 elif answer_type == FHIRAnswerType.INTEGER:

817 fhir_answer = int(raw_answer) # type: ignore[assignment]

818 elif answer_type == FHIRAnswerType.QUANTITY:

819 fhir_answer = Quantity(

820 jsondict={

821 Fc.VALUE: float(raw_answer)

822 # More sophistication is possible -- units, for example.

823 }

824 ).as_json()

825 elif answer_type == FHIRAnswerType.STRING:

826 fhir_answer = str(raw_answer) # type: ignore[assignment]

827 elif answer_type == FHIRAnswerType.TIME:

828 fhir_answer = FHIRDate(

829 format_datetime(raw_answer, DateFormat.FHIR_TIME)

830 ).as_json()

831 elif answer_type == FHIRAnswerType.URI:

832 fhir_answer = str(raw_answer) # type: ignore[assignment]

833 else:

834 raise NotImplementedError(

835 f"Don't know how to handle FHIR answer type {answer_type}"

836 )

837

838 # Build the FHIR object

839 return QuestionnaireResponseItemAnswer(

840 jsondict={answer_type.value: fhir_answer}

841 )

842

843 def questionnaire_response_item(self) -> Dict:

844 """

845 Returns a JSON/dict representation of a FHIR QuestionnaireResponseItem.

846 """

847 answer = self._qr_item_answer()

848 return QuestionnaireResponseItem(

849 jsondict={

850 Fc.LINK_ID: self.qname,

851 Fc.TEXT: self.qtext, # question text

852 Fc.ANSWER: [answer.as_json()],

853 # Not supported yet: nesting, via "item".

854 }

855 ).as_json()

Coverage for cc_modules/cc_fhir.py: 43%

175 statements