Add Devices Spec

care/emr/api/viewsets/device.py

@@ -0,0 +1,223 @@
+from django.db import transaction
+from django.utils import timezone
+from django_filters import rest_framework as filters
+from pydantic import UUID4, BaseModel
+from rest_framework.decorators import action
+from rest_framework.exceptions import PermissionDenied, ValidationError
+from rest_framework.generics import get_object_or_404
+from rest_framework.response import Response
+
+from care.emr.api.viewsets.base import EMRModelReadOnlyViewSet, EMRModelViewSet
+from care.emr.models import (
+    Device,
+    DeviceEncounterHistory,
+    DeviceLocationHistory,
+    Encounter,
+    FacilityLocation,
+)
+from care.emr.models.organization import FacilityOrganizationUser
+from care.emr.resources.device.spec import (
+    DeviceCreateSpec,
+    DeviceEncounterHistoryListSpec,
+    DeviceListSpec,
+    DeviceLocationHistoryListSpec,
+    DeviceRetrieveSpec,
+    DeviceUpdateSpec,
+)
+from care.facility.models import Facility
+from care.security.authorization import AuthorizationController
+
+
+class DeviceFilters(filters.FilterSet):
+    current_location = filters.UUIDFilter(field_name="current_location__external_id")
+    current_encounter = filters.UUIDFilter(field_name="current_encounter__external_id")
+
+
+class DeviceViewSet(EMRModelViewSet):
+    database_model = Device
+    pydantic_model = DeviceCreateSpec
+    pydantic_update_model = DeviceUpdateSpec
+    pydantic_read_model = DeviceListSpec
+    pydantic_retrieve_model = DeviceRetrieveSpec
+    filterset_class = DeviceFilters
+    filter_backends = [filters.DjangoFilterBackend]
+
+    def get_facility_obj(self):
+        return get_object_or_404(
+            Facility, external_id=self.kwargs["facility_external_id"]
+        )
+
+    def authorize_create(self, instance):
+        if not AuthorizationController.call("can_manage_devices", self.request.user):
+            raise PermissionDenied("You do not have permission to create device")
+
+    def authorize_update(self, instance, model_instance):
+        if not AuthorizationController.call("can_manage_devices", self.request.user):
+            raise PermissionDenied("You do not have permission to update device")
+
+    def authorize_destroy(self, instance):
+        if not AuthorizationController.call("can_manage_devices", self.request.user):
+            raise PermissionDenied("You do not have permission to delete device")
+
+    def perform_create(self, instance):
+        instance.facility = self.get_facility_obj()
+        super().perform_create(instance)
+
+    def get_queryset(self):
+        """
+        When Location is specified, Location permission is checked (or) organization filters are applied
+        If location is not specified the organization cache is used
+        """
+        queryset = Device.objects.all()
+
+        if self.request.user.is_superuser:
+            return queryset
+
+        facility = self.get_facility_obj()
+
+        users_facility_organizations = FacilityOrganizationUser.objects.filter(
+            organization__facility=facility, user=self.request.user
+        ).values_list("organization_id", flat=True)
+
+        if "location" in self.request.GET:
+            queryset = queryset.filter(
+                facility_organization_cache__overlap=users_facility_organizations
+            )
+            # TODO Check access to location with permission and then allow filter with or,
+            # Access should not be limited by location if the device has org access
+            # If location access then allow all, otherwise apply organization filter
+        else:
+            queryset = queryset.filter(
+                facility_organization_cache__overlap=users_facility_organizations
+            )
+
+        return queryset
+
+    class DeviceEncounterAssociationRequest(BaseModel):
+        encounter: UUID4 | None = None
+
+    @action(detail=True, methods=["POST"])
+    def associate_encounter(self, request, *args, **kwargs):
+        request_data = self.DeviceEncounterAssociationRequest(**request.data)
+        encounter = None
+        if request_data.encounter:
+            encounter = get_object_or_404(Encounter, external_id=request_data.encounter)
+        device = self.get_object()
+        facility = self.get_facility_obj()
+
+        if not AuthorizationController.call(
+            "can_associate_device_encounter", self.request.user, facility
+        ):
+            raise PermissionDenied(
+                "You do not have permission to associate encounter to this device"
+            )
+
+        if encounter and device.current_encounter_id == encounter.id:
+            raise ValidationError("Encounter already associated")
+        if encounter and encounter.facility_id != facility.id:
+            raise ValidationError("Encounter is not part of given facility")
+        with transaction.atomic():
+            if device.current_encounter:
+                old_obj = DeviceEncounterHistory.objects.filter(
+                    device=device, encounter=device.current_encounter, end__isnull=True
+                ).first()
+                if old_obj:
+                    old_obj.end = timezone.now()
+                    old_obj.save()
+            device.current_encounter = encounter
+            device.save(update_fields=["current_encounter"])
+            if encounter:
+                obj = DeviceEncounterHistory.objects.create(
+                    device=device,
+                    encounter=encounter,
+                    start=timezone.now(),
+                    created_by=request.user,
+                )
+                return Response(DeviceEncounterHistoryListSpec.serialize(obj).to_json())
+            return Response({})
+
+    class DeviceLocationAssociationRequest(BaseModel):
+        location: UUID4 | None = None
+
+    @action(detail=True, methods=["POST"])
+    def associate_location(self, request, *args, **kwargs):
+        request_data = self.DeviceLocationAssociationRequest(**request.data)
+        location = None
+        if request_data.location:
+            location = get_object_or_404(
+                FacilityLocation, external_id=request_data.location
+            )
+        facility = self.get_facility_obj()
+        device = self.get_object()
+
+        if not AuthorizationController.call(
+            "can_associate_device_location", self.request.user, facility
+        ):
+            raise PermissionDenied(
+                "You do not have permission to associate location to this device"
+            )
+        if location and device.current_location_id == location.id:
+            raise ValidationError("Location already associated")
+        if location and location.facility_id != facility.id:
+            raise ValidationError("Location is not part of given facility")
+        with transaction.atomic():
+            if device.current_location:
+                old_obj = DeviceLocationHistory.objects.filter(
+                    device=device, location=device.current_location, end__isnull=True
+                ).first()
+                if old_obj:
+                    old_obj.end = timezone.now()
+                    old_obj.save()
+            device.current_location = location
+            device.save(update_fields=["current_location"])
+            if location:
+                obj = DeviceLocationHistory.objects.create(
+                    device=device,
+                    location=location,
+                    start=timezone.now(),
+                    created_by=request.user,
+                )
+                return Response(DeviceLocationHistoryListSpec.serialize(obj).to_json())
+            return Response({})
+
+
+class DeviceLocationHistoryViewSet(EMRModelReadOnlyViewSet):
+    database_model = DeviceLocationHistory
+    pydantic_read_model = DeviceLocationHistoryListSpec
+
+    def get_device(self):
+        return get_object_or_404(Device, external_id=self.kwargs["device_external_id"])
+
+    def get_queryset(self):
+        if not AuthorizationController.call(
+            "can_list_devices",
+            self.request.user,
+        ):
+            raise PermissionDenied("You do not have permission to access the device")
+
+        return (
+            DeviceLocationHistory.objects.filter(device=self.get_device())
+            .select_related("location")
+            .order_by("-end")
+        )
+
+
+class DeviceEncounterHistoryViewSet(EMRModelReadOnlyViewSet):
+    database_model = DeviceEncounterHistory
+    pydantic_read_model = DeviceEncounterHistoryListSpec
+
+    def get_device(self):
+        return get_object_or_404(Device, external_id=self.kwargs["device_external_id"])
+
+    def get_queryset(self):
+        if not AuthorizationController.call(
+            "can_list_devices",
+            self.request.user,
+        ):
+            raise PermissionDenied("You do not have permission to access the device")
+
+        return (
+            DeviceEncounterHistory.objects.filter(device=self.get_device())
+            .select_related("encounter", "encounter__patient", "encounter__facility")
+            .order_by("-end")
+        )

care/emr/api/viewsets/encounter.py

@@ -20,6 +20,8 @@
     EMRUpdateMixin,
 )
 from care.emr.models import (
+    Device,
+    DeviceEncounterHistory,
     Encounter,
     EncounterOrganization,
     FacilityOrganization,
@@ -96,6 +98,22 @@ def perform_create(self, instance):
             if not organizations:
                 instance.sync_organization_cache()
 
+    def perform_update(self, instance):
+        with transaction.atomic():
+            if instance.status in COMPLETED_CHOICES:
+                device_ids = list(
+                    Device.objects.filter(current_encounter=instance).values_list(
+                        "id", flat=True
+                    )
+                )
+                Device.objects.filter(id__in=device_ids).update(current_encounter=None)
+
+                DeviceEncounterHistory.objects.filter(
+                    device_id__in=device_ids, encounter=instance, end__isnull=True
+                ).update(end=timezone.now())
+
+            super().perform_update(instance)
+
     def authorize_update(self, request_obj, model_instance):
         if not AuthorizationController.call(
             "can_update_encounter_obj", self.request.user, model_instance

care/emr/migrations/0017_device_deviceencounterhistory_devicelocationhistory_and_more.py

@@ -0,0 +1,118 @@
+# Generated by Django 5.1.4 on 2025-02-06 14:08
+
+import django.contrib.postgres.fields
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('emr', '0016_allergyintolerance_copied_from'),
+        ('facility', '0476_facility_default_internal_organization_and_more'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Device',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('external_id', models.UUIDField(db_index=True, default=uuid.uuid4, unique=True)),
+                ('created_date', models.DateTimeField(auto_now_add=True, db_index=True, null=True)),
+                ('modified_date', models.DateTimeField(auto_now=True, db_index=True, null=True)),
+                ('deleted', models.BooleanField(db_index=True, default=False)),
+                ('history', models.JSONField(default=dict)),
+                ('meta', models.JSONField(default=dict)),
+                ('identifier', models.CharField(blank=True, max_length=1024, null=True)),
+                ('status', models.CharField(max_length=14)),
+                ('availability_status', models.CharField(max_length=14)),
+                ('manufacturer', models.CharField(max_length=1024)),
+                ('manufacture_date', models.DateTimeField(blank=True, null=True)),
+                ('expiration_date', models.DateTimeField(blank=True, null=True)),
+                ('lot_number', models.CharField(blank=True, max_length=1024, null=True)),
+                ('serial_number', models.CharField(blank=True, max_length=1024, null=True)),
+                ('registered_name', models.CharField(blank=True, max_length=1024, null=True)),
+                ('user_friendly_name', models.CharField(blank=True, max_length=1024, null=True)),
+                ('model_number', models.CharField(blank=True, max_length=1024, null=True)),
+                ('part_number', models.CharField(blank=True, max_length=1024, null=True)),
+                ('contact', models.JSONField(default=dict)),
+                ('care_type', models.CharField(blank=True, max_length=1024, null=True)),
+                ('facility_organization_cache', django.contrib.postgres.fields.ArrayField(base_field=models.IntegerField(), default=list, size=None)),
+                ('created_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_created_by', to=settings.AUTH_USER_MODEL)),
+                ('current_encounter', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='emr.encounter')),
+                ('current_location', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='emr.facilitylocation')),
+                ('facility', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='facility.facility')),
+                ('managing_organization', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='emr.facilityorganization')),
+                ('updated_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_updated_by', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='DeviceEncounterHistory',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('external_id', models.UUIDField(db_index=True, default=uuid.uuid4, unique=True)),
+                ('created_date', models.DateTimeField(auto_now_add=True, db_index=True, null=True)),
+                ('modified_date', models.DateTimeField(auto_now=True, db_index=True, null=True)),
+                ('deleted', models.BooleanField(db_index=True, default=False)),
+                ('history', models.JSONField(default=dict)),
+                ('meta', models.JSONField(default=dict)),
+                ('start', models.DateTimeField()),
+                ('end', models.DateTimeField(blank=True, null=True)),
+                ('created_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_created_by', to=settings.AUTH_USER_MODEL)),
+                ('device', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='emr.device')),
+                ('encounter', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='emr.encounter')),
+                ('updated_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_updated_by', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='DeviceLocationHistory',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('external_id', models.UUIDField(db_index=True, default=uuid.uuid4, unique=True)),
+                ('created_date', models.DateTimeField(auto_now_add=True, db_index=True, null=True)),
+                ('modified_date', models.DateTimeField(auto_now=True, db_index=True, null=True)),
+                ('deleted', models.BooleanField(db_index=True, default=False)),
+                ('history', models.JSONField(default=dict)),
+                ('meta', models.JSONField(default=dict)),
+                ('start', models.DateTimeField()),
+                ('end', models.DateTimeField(blank=True, null=True)),
+                ('created_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_created_by', to=settings.AUTH_USER_MODEL)),
+                ('device', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='emr.device')),
+                ('location', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='emr.facilitylocation')),
+                ('updated_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_updated_by', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='DeviceServiceHistory',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('external_id', models.UUIDField(db_index=True, default=uuid.uuid4, unique=True)),
+                ('created_date', models.DateTimeField(auto_now_add=True, db_index=True, null=True)),
+                ('modified_date', models.DateTimeField(auto_now=True, db_index=True, null=True)),
+                ('deleted', models.BooleanField(db_index=True, default=False)),
+                ('history', models.JSONField(default=dict)),
+                ('meta', models.JSONField(default=dict)),
+                ('serviced_on', models.DateField(default=None, null=True)),
+                ('note', models.TextField(blank=True, default='', null=True)),
+                ('edit_history', models.JSONField(default=list)),
+                ('created_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_created_by', to=settings.AUTH_USER_MODEL)),
+                ('device', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='emr.device')),
+                ('updated_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_updated_by', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

care/emr/migrations/0018_alter_device_care_type_alter_device_status.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.1.4 on 2025-02-11 10:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('emr', '0017_device_deviceencounterhistory_devicelocationhistory_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='device',
+            name='care_type',
+            field=models.CharField(blank=True, default=None, max_length=1024, null=True),
+        ),
+        migrations.AlterField(
+            model_name='device',
+            name='status',
+            field=models.CharField(max_length=16),
+        ),
+    ]