Add Devices Spec

care/emr/api/viewsets/device.py

@@ -0,0 +1,163 @@
+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 ValidationError
+from rest_framework.generics import get_object_or_404
+
+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
+
+
+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 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
+            # 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
+
+    @action(detail=True, methods=["POST"])
+    def associate_encounter(self, request, *args, **kwargs):
+        request_data = self.DeviceEncounterAssociationRequest(**request.data)
+        encounter = get_object_or_404(Encounter, external_id=request_data.encounter)
+        device = self.get_object()
+        # TODO Perform Authz for encounter
+        if device.current_encounter_id == encounter.id:
+            raise ValidationError("Encounter already associated")
+        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"])
+            DeviceEncounterHistory.objects.create(
+                device=device, encounter=encounter, start=timezone.now()
+            )
+
+    class DeviceLocationAssociationRequest(BaseModel):
+        location: UUID4
+
+    @action(detail=True, methods=["POST"])
+    def associate_location(self, request, *args, **kwargs):
+        request_data = self.DeviceLocationAssociationRequest(**request.data)
+        location = get_object_or_404(
+            FacilityLocation, external_id=request_data.location
+        )
+        device = self.get_object()
+        # TODO Perform Authz for location
+        if device.current_location == location.id:
+            raise ValidationError("Location already associated")
+        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"])
+            DeviceLocationHistory.objects.create(
+                device=device, location=location, start=timezone.now()
+            )
+
+
+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):
+        return DeviceLocationHistory.objects.filter(
+            device=self.get_device()
+        ).select_related("location")
+
+    # TODO Authz
+
+
+class DeviceEncounterHistoryViewSet(EMRModelReadOnlyViewSet):
+    database_model = DeviceLocationHistory
+    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):
+        return DeviceLocationHistory.objects.filter(
+            device=self.get_device()
+        ).select_related("encounter")
+
+
+# TODO AuthZ
+# TODO Serialize current location and history in the retrieve API

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/models/__init__.py

@@ -8,3 +8,4 @@
 from .patient import *  # noqa F403
 from .file_upload import *  # noqa F403
 from .location import *  # noqa F403
+from .device import *  # noqa F403

care/emr/models/device.py

@@ -0,0 +1,79 @@
+from django.contrib.postgres.fields import ArrayField
+from django.db import models
+
+from care.emr.models import EMRBaseModel
+
+
+class Device(EMRBaseModel):
+    # Device Data
+    identifier = models.CharField(max_length=1024, null=True, blank=True)
+    status = models.CharField(max_length=14)
+    availability_status = models.CharField(max_length=14)
+    manufacturer = models.CharField(max_length=1024)
+    manufacture_date = models.DateTimeField(null=True, blank=True)
+    expiration_date = models.DateTimeField(null=True, blank=True)
+    lot_number = models.CharField(max_length=1024, null=True, blank=True)
+    serial_number = models.CharField(max_length=1024, null=True, blank=True)
+    registered_name = models.CharField(max_length=1024, null=True, blank=True)
+    user_friendly_name = models.CharField(max_length=1024, null=True, blank=True)
+    model_number = models.CharField(max_length=1024, null=True, blank=True)
+    part_number = models.CharField(max_length=1024, null=True, blank=True)
+    contact = models.JSONField(default=dict)
+    care_type = models.CharField(max_length=1024, null=True, blank=True, default=None)
+
+    # Relations
+    facility = models.ForeignKey("facility.Facility", on_delete=models.CASCADE)
+    managing_organization = models.ForeignKey(
+        "emr.FacilityOrganization", on_delete=models.SET_NULL, null=True, blank=True
+    )
+    current_location = models.ForeignKey(
+        "emr.FacilityLocation", on_delete=models.SET_NULL, null=True, blank=True
+    )
+    current_encounter = models.ForeignKey(
+        "emr.Encounter", on_delete=models.SET_NULL, null=True, blank=True
+    )
+
+    # metadata
+    facility_organization_cache = ArrayField(models.IntegerField(), default=list)
+
+    def save(self, *args, **kwargs):
+        from care.emr.models.organization import FacilityOrganization
+
+        facility_root_org = FacilityOrganization.objects.filter(
+            org_type="root", facility=self.facility
+        ).first()
+        orgs = set()
+        if facility_root_org:
+            orgs = orgs.union({facility_root_org.id})
+        if self.managing_organization:
+            orgs = orgs.union(
+                {
+                    *self.managing_organization.parent_cache,
+                    self.managing_organization.id,
+                }
+            )
+        self.facility_organization_cache = list(orgs)
+        return super().save(*args, **kwargs)
+
+
+class DeviceEncounterHistory(EMRBaseModel):
+    device = models.ForeignKey("emr.Device", on_delete=models.CASCADE)
+    encounter = models.ForeignKey("emr.Encounter", on_delete=models.CASCADE)
+    start = models.DateTimeField()
+    end = models.DateTimeField(null=True, blank=True)
+
+
+class DeviceLocationHistory(EMRBaseModel):
+    device = models.ForeignKey("emr.Device", on_delete=models.CASCADE)
+    location = models.ForeignKey("emr.FacilityLocation", on_delete=models.CASCADE)
+    start = models.DateTimeField()
+    end = models.DateTimeField(null=True, blank=True)
+
+
+class DeviceServiceHistory(EMRBaseModel):
+    device = models.ForeignKey(
+        Device, on_delete=models.PROTECT, null=False, blank=False
+    )
+    serviced_on = models.DateField(default=None, null=True, blank=False)
+    note = models.TextField(default="", null=True, blank=True)
+    edit_history = models.JSONField(default=list)