Labs

care/emr/api/viewsets/diagnostic_report.py

@@ -0,0 +1,140 @@
+from datetime import UTC, datetime
+
+from django_filters import CharFilter, FilterSet, OrderingFilter, UUIDFilter
+from django_filters.rest_framework import DjangoFilterBackend
+from drf_spectacular.utils import extend_schema
+from rest_framework.decorators import action
+from rest_framework.response import Response
+
+from care.emr.api.viewsets.base import EMRModelViewSet
+from care.emr.models.diagnostic_report import DiagnosticReport
+from care.emr.models.observation import Observation
+from care.emr.resources.diagnostic_report.spec import (
+    DiagnosticReportCreateSpec,
+    DiagnosticReportListSpec,
+    DiagnosticReportObservationRequest,
+    DiagnosticReportRetrieveSpec,
+    DiagnosticReportReviewRequest,
+    DiagnosticReportUpdateSpec,
+    DiagnosticReportVerifyRequest,
+    StatusChoices,
+)
+from care.emr.resources.observation.spec import Performer, PerformerType
+
+
+class DiagnosticReportFilters(FilterSet):
+    phase = CharFilter(field_name="based_on__phase", lookup_expr="iexact")
+    status = CharFilter(field_name="status", lookup_expr="iexact")
+    specimen = UUIDFilter(field_name="specimen__external_id")
+    based_on = UUIDFilter(field_name="based_on__external_id")
+
+    ordering = OrderingFilter(
+        fields=(
+            ("created_date", "created_date"),
+            ("modified_date", "modified_date"),
+        )
+    )
+
+
+class DiagnosticReportViewSet(EMRModelViewSet):
+    database_model = DiagnosticReport
+    pydantic_model = DiagnosticReportCreateSpec
+    pydantic_update_model = DiagnosticReportUpdateSpec
+    pydantic_read_model = DiagnosticReportListSpec
+    pydantic_retrieve_model = DiagnosticReportRetrieveSpec
+    filter_backends = [DjangoFilterBackend]
+    filterset_class = DiagnosticReportFilters
+
+    def clean_create_data(self, request, *args, **kwargs):
+        clean_data = super().clean_create_data(request, *args, **kwargs)
+
+        clean_data["performer"] = self.request.user.external_id
+        return clean_data
+
+    @extend_schema(
+        request=DiagnosticReportObservationRequest,
+        responses={200: DiagnosticReportRetrieveSpec},
+        tags=["diagnostic_report"],
+    )
+    @action(detail=True, methods=["POST"])
+    def observations(self, request, *args, **kwargs):
+        data = DiagnosticReportObservationRequest(**request.data)
+        report: DiagnosticReport = self.get_object()
+
+        observations = []
+        for observation in data.observations:
+            if not observation.performer:
+                observation.performer = Performer(
+                    type=PerformerType.user,
+                    id=str(request.user.external_id),
+                )
+
+            observation_instance = observation.de_serialize()
+            observation_instance.subject_id = report.subject.id
+            observation_instance.encounter = report.encounter
+            observation_instance.patient = report.subject
+
+            observations.append(observation_instance)
+
+        observation_instances = Observation.objects.bulk_create(observations)
+        report.result.set(observation_instances)
+        report.status = StatusChoices.partial
+        report.save()
+
+        return Response(
+            self.get_retrieve_pydantic_model().serialize(report).to_json(),
+        )
+
+    @extend_schema(
+        request=DiagnosticReportVerifyRequest,
+        responses={200: DiagnosticReportRetrieveSpec},
+        tags=["diagnostic_report"],
+    )
+    @action(detail=True, methods=["POST"])
+    def verify(self, request, *args, **kwargs):
+        data = DiagnosticReportVerifyRequest(**request.data)
+        report: DiagnosticReport = self.get_object()
+
+        if data.is_approved:
+            report.status = StatusChoices.preliminary
+        else:
+            report.status = StatusChoices.cancelled
+
+        report.issued = datetime.now(UTC)
+        report.save()
+
+        return Response(
+            self.get_retrieve_pydantic_model().serialize(report).to_json(),
+        )
+
+    @extend_schema(
+        request=DiagnosticReportReviewRequest,
+        responses={200: DiagnosticReportRetrieveSpec},
+        tags=["diagnostic_report"],
+    )
+    @action(detail=True, methods=["POST"])
+    def review(self, request, *args, **kwargs):
+        data = DiagnosticReportReviewRequest(**request.data)
+        report: DiagnosticReport = self.get_object()
+
+        if (
+            report.results_interpreter
+            and report.results_interpreter.external_id != request.user.external_id
+        ):
+            return Response(
+                {"detail": "This report is assigned to a different user for review."},
+                status=403,
+            )
+
+        if data.is_approved:
+            report.status = StatusChoices.final
+        else:
+            report.status = StatusChoices.cancelled
+
+        report.conclusion = data.conclusion
+        report.results_interpreter = request.user
+        report.save()
+
+        return Response(
+            self.get_retrieve_pydantic_model().serialize(report).to_json(),
+        )

care/emr/api/viewsets/service_request.py

@@ -0,0 +1,40 @@
+from django_filters import CharFilter, FilterSet, OrderingFilter, UUIDFilter
+from django_filters.rest_framework import DjangoFilterBackend
+
+from care.emr.api.viewsets.base import EMRModelViewSet
+from care.emr.models.service_request import ServiceRequest
+from care.emr.resources.service_request.spec import (
+    ServiceRequestCreateSpec,
+    ServiceRequestListSpec,
+    ServiceRequestRetrieveSpec,
+    ServiceRequestUpdateSpec,
+)
+
+
+class ServiceRequestFilters(FilterSet):
+    phase = CharFilter(field_name="phase", lookup_expr="iexact")
+    subject = UUIDFilter(field_name="subject__external_id")
+    encounter = UUIDFilter(field_name="encounter__external_id")
+
+    ordering = OrderingFilter(
+        fields=(
+            ("created_date", "created_date"),
+            ("modified_date", "modified_date"),
+        )
+    )
+
+
+class ServiceRequestViewSet(EMRModelViewSet):
+    database_model = ServiceRequest
+    pydantic_model = ServiceRequestCreateSpec
+    pydantic_update_model = ServiceRequestUpdateSpec
+    pydantic_read_model = ServiceRequestListSpec
+    pydantic_retrieve_model = ServiceRequestRetrieveSpec
+    filter_backends = [DjangoFilterBackend]
+    filterset_class = ServiceRequestFilters
+
+    def clean_create_data(self, request, *args, **kwargs):
+        clean_data = super().clean_create_data(request, *args, **kwargs)
+
+        clean_data["requester"] = self.request.user.external_id
+        return clean_data

care/emr/api/viewsets/specimen.py

@@ -0,0 +1,143 @@
+from datetime import UTC, datetime
+
+from django.db.models import Q
+from django_filters import CharFilter, FilterSet, OrderingFilter, UUIDFilter
+from django_filters.rest_framework import DjangoFilterBackend
+from drf_spectacular.utils import extend_schema
+from rest_framework.decorators import action
+from rest_framework.generics import get_object_or_404
+from rest_framework.response import Response
+
+from care.emr.api.viewsets.base import EMRModelViewSet
+from care.emr.models.specimen import Specimen
+from care.emr.resources.specimen.spec import (
+    SpecimenCollectRequest,
+    SpecimenCreateSpec,
+    SpecimenListSpec,
+    SpecimenProcessRequest,
+    SpecimenReceiveAtLabRequest,
+    SpecimenRetrieveSpec,
+    SpecimenSendToLabRequest,
+    SpecimenUpdateSpec,
+    StatusChoices,
+)
+
+
+class SpecimenFilters(FilterSet):
+    phase = CharFilter(field_name="request__phase", lookup_expr="iexact")
+    request = UUIDFilter(field_name="request__external_id")
+    encounter = UUIDFilter(field_name="request__encounter__external_id")
+
+    ordering = OrderingFilter(
+        fields=(
+            ("created_date", "created_date"),
+            ("modified_date", "modified_date"),
+        )
+    )
+
+
+class SpecimenViewSet(EMRModelViewSet):
+    database_model = Specimen
+    pydantic_model = SpecimenCreateSpec
+    pydantic_update_model = SpecimenUpdateSpec
+    pydantic_read_model = SpecimenListSpec
+    pydantic_retrieve_model = SpecimenRetrieveSpec
+    filter_backends = [DjangoFilterBackend]
+    filterset_class = SpecimenFilters
+
+    def get_object(self) -> Specimen:
+        return get_object_or_404(
+            self.get_queryset(),
+            Q(external_id__iexact=self.kwargs[self.lookup_field])
+            | Q(identifier=self.kwargs[self.lookup_field])
+            | Q(accession_identifier=self.kwargs[self.lookup_field]),
+        )
+
+    @extend_schema(
+        request=SpecimenCollectRequest,
+        responses={200: SpecimenRetrieveSpec},
+        tags=["specimen"],
+    )
+    @action(detail=True, methods=["POST"])
+    def collect(self, request, *args, **kwargs):
+        data = SpecimenCollectRequest(**request.data)
+        specimen = self.get_object()
+
+        specimen.identifier = data.identifier
+        specimen.status = StatusChoices.available
+        specimen.collected_at = datetime.now(UTC)
+        specimen.collected_by = request.user
+        specimen.save()
+
+        return Response(
+            self.get_retrieve_pydantic_model().serialize(specimen).to_json(),
+        )
+
+    @extend_schema(
+        request=SpecimenSendToLabRequest,
+        responses={200: SpecimenRetrieveSpec},
+        tags=["specimen"],
+    )
+    @action(detail=True, methods=["POST"])
+    def send_to_lab(self, request, *args, **kwargs):
+        data = SpecimenSendToLabRequest(**request.data)
+        specimen = self.get_object()
+        service_request = specimen.request
+
+        service_request.location = data.lab
+        specimen.dispatched_at = datetime.now(UTC)
+        specimen.dispatched_by = request.user
+        service_request.save()
+        specimen.save()
+
+        return Response(
+            self.get_retrieve_pydantic_model().serialize(specimen).to_json(),
+        )
+
+    @extend_schema(
+        request=SpecimenReceiveAtLabRequest,
+        responses={200: SpecimenRetrieveSpec},
+        tags=["specimen"],
+    )
+    @action(detail=True, methods=["POST"])
+    def receive_at_lab(self, request, *args, **kwargs):
+        data = SpecimenReceiveAtLabRequest(**request.data)
+        specimen = self.get_object()
+
+        specimen.accession_identifier = data.accession_identifier
+        specimen.condition = data.condition
+        specimen.received_at = datetime.now(UTC)
+        specimen.received_by = request.user
+        specimen.note = data.note
+        specimen.save()
+
+        return Response(
+            self.get_retrieve_pydantic_model().serialize(specimen).to_json(),
+        )
+
+    @extend_schema(
+        request=SpecimenProcessRequest,
+        responses={200: SpecimenRetrieveSpec},
+        tags=["specimen"],
+    )
+    @action(detail=True, methods=["POST"])
+    def process(self, request, *args, **kwargs):
+        data = SpecimenProcessRequest(**request.data)
+        specimen = self.get_object()
+
+        processes = []
+        for process in data.process:
+            if not process.time:
+                process.time = datetime.now(UTC)
+
+            if not process.performer:
+                process.performer = request.user.external_id
+
+            processes.append(process.model_dump(mode="json"))
+
+        specimen.processing.extend(processes)
+        specimen.save()
+
+        return Response(
+            self.get_retrieve_pydantic_model().serialize(specimen).to_json(),
+        )

care/emr/apps.py

@@ -5,3 +5,6 @@
 class EMRConfig(AppConfig):
     name = "care.emr"
     verbose_name = _("Electronic Medical Record")
+
+    def ready(self):
+        import care.emr.signals  # noqa