Add Devices Spec

care/emr/api/viewsets/device.py

@@ -0,0 +1,70 @@
+from django_filters import rest_framework as filters
+from rest_framework.generics import get_object_or_404
+
+from care.emr.api.viewsets.base import EMRModelViewSet
+from care.emr.models import Device
+from care.emr.models.organization import FacilityOrganizationUser
+from care.emr.resources.device.spec import (
+    DeviceCreateSpec,
+    DeviceListSpec,
+    DeviceRetrieveSpec,
+    DeviceUpdateSpec,
+)
+from care.facility.models import Facility
+
+
+class DeviceFilters(filters.FilterSet):
+    pass
+
+
+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
+
+    # TODO Action for Associating Encounter
+    # TODO Action for Associating Location
+    # TODO RO API's for Device Location and Encounter History
+    # 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)

care/emr/registries/device_type/device_registry.py

@@ -0,0 +1,50 @@
+class DeviceTypeBase:
+    def handle_create(self, request_data, obj):
+        """
+        Handle Creates, the original source request along with the base object created is passed along.
+        Update the obj as needed and create any extra metadata needed. This method is called within a transaction
+        """
+        return obj
+
+    def handle_update(self, request_data, obj):
+        """
+        Handle Updates, the original source request along with the base object updated is passed along.
+        Update the obj as needed and create any extra metadata needed. This method is called within a transaction
+        """
+        return obj
+
+    def handle_delete(self, obj):
+        """
+        Handle Deletes, the object to be deleted is passed along.
+        Perform validation or any other changes required here
+        """
+        return obj
+
+    def list(self, obj):
+        """
+        Return Extra metadata for the given obj for lists, N+1 queries is okay, caching is recommended for performance
+        """
+        return {}
+
+    def retrieve(self, obj):
+        """
+        Return Extra metadata for the given obj during retrieves
+        """
+        return {}
+
+    def perform_action(self, obj, action, request):
+        """
+        Perform some kind of action on an asset, the HTTP request is proxied through as is.
+        an HTTP response object is expected as the return.
+        """
+        return  # Return an HTTP Response
+
+
+class DeviceTypeRegistry:
+    _device_types = {}
+
+    @classmethod
+    def register(cls, device_type, device_class) -> None:
+        if not issubclass(device_class, DeviceTypeBase):
+            raise ValueError("The provided class is not a subclass of DeviceTypeBase")
+        cls._device_types[device_type] = device_class

care/emr/resources/common/contact_point.py

@@ -0,0 +1,27 @@
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class ContactPointSystemChoices(str, Enum):
+    phone = "phone"
+    fax = "fax"
+    email = "email"
+    pager = "pager"
+    url = "url"
+    sms = "sms"
+    other = "other"
+
+
+class ContactPointUseChoices(str, Enum):
+    home = "home"
+    work = "work"
+    temp = "temp"
+    old = "old"
+    mobile = "mobile"
+
+
+class ContactPoint(BaseModel):
+    system: ContactPointSystemChoices
+    value: str
+    use: ContactPointUseChoices