Patient privacy feature

care/facility/api/viewsets/asset.py

@@ -334,9 +334,10 @@ def operate_assets(self, request, *args, **kwargs):
                     "middleware_hostname": middleware_hostname,
                 }
             )
+
+            asset_class.validate_action(action)
             result = asset_class.handle_action(action)
             return Response({"result": result}, status=status.HTTP_200_OK)
-
         except ValidationError as e:
             return Response({"message": e.detail}, status=status.HTTP_400_BAD_REQUEST)
 

care/facility/api/viewsets/bed.py

@@ -4,6 +4,7 @@
 from drf_spectacular.utils import extend_schema, extend_schema_view
 from rest_framework import filters as drf_filters
 from rest_framework import status
+from rest_framework.decorators import action
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.exceptions import ValidationError as DRFValidationError
 from rest_framework.fields import get_error_detail
@@ -232,3 +233,44 @@
             allowed_facilities = get_accessible_facilities(user)
             queryset = queryset.filter(bed__facility__id__in=allowed_facilities)
         return queryset
+
+    @extend_schema(
+        description="Toggle patient privacy",
+        responses={status.HTTP_200_OK: None},
+        request=None,
+        tags=["consultationbed"],
+    )
+    @action(detail=True, methods=["PATCH"])
+    def toggle_patient_privacy(self, request, external_id):
+        try:
+            user: User = request.user
+            consultation_bed: ConsultationBed = (
+                self.get_queryset().filter(external_id=external_id).first()
+            )
+
+            if consultation_bed and (
+                user.user_type == User.TYPE_VALUE_MAP["WardAdmin"]
+                or user.user_type == User.TYPE_VALUE_MAP["LocalBodyAdmin"]
+                or user.user_type == User.TYPE_VALUE_MAP["DistrictAdmin"]
+                or user.user_type == User.TYPE_VALUE_MAP["StateAdmin"]
+                or (
+                    user.user_type == User.TYPE_VALUE_MAP["Doctor"]
+                    and user.home_facility.external_id
+                    == consultation_bed.bed.facility.external_id
+                )
+                or (
+                    user.user_type == User.TYPE_VALUE_MAP["Staff"]
+                    and user.home_facility.external_id
+                    == consultation_bed.bed.facility.external_id
+                )
+            ):
+                consultation_bed.privacy = not consultation_bed.privacy
+                consultation_bed.save()
+                return Response({"status": "success"}, status=status.HTTP_200_OK)
+            raise PermissionDenied(
+                detail="You do not have permission to perform this action"
+            )
+        except PermissionDenied as e:
+            return Response({"message": e.detail}, status=status.HTTP_403_FORBIDDEN)
+        except Exception as e:
+            return Response({"message": str(e)}, status=status.HTTP_400_BAD_REQUEST)

care/facility/migrations/0397_consultationbed_privacy.py

@@ -0,0 +1,17 @@
+# Generated by Django 4.2.7 on 2023-11-23 13:22
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("facility", "0396_merge_20231122_0240"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="consultationbed",
+            name="privacy",
+            field=models.BooleanField(default=False),
+        ),
+    ]

care/facility/models/bed.py

@@ -72,6 +72,7 @@ class ConsultationBed(BaseModel):
     bed = models.ForeignKey(Bed, on_delete=models.PROTECT, null=False, blank=False)
     start_date = models.DateTimeField(null=False, blank=False)
     end_date = models.DateTimeField(null=True, blank=True, default=None)
+    privacy = models.BooleanField(default=False)
     meta = JSONField(default=dict, blank=True)
     assets = models.ManyToManyField(
         Asset, through="ConsultationBedAsset", related_name="assigned_consultation_beds"

care/facility/tests/test_asset_operate_api.py

@@ -0,0 +1,103 @@
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from care.facility.models import AssetBed
+from care.utils.tests.test_utils import TestUtils
+
+
+class AssetViewSetTestCase(TestUtils, APITestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.state = cls.create_state()
+        cls.district = cls.create_district(state=cls.state)
+        cls.local_body = cls.create_local_body(cls.district)
+        cls.user = cls.create_user(district=cls.district, username="test user")
+        cls.facility = cls.create_facility(
+            district=cls.district, local_body=cls.local_body, user=cls.user
+        )
+        cls.asset1_location = cls.create_asset_location(facility=cls.facility)
+
+        # depends upon the operational dev camera config
+        cls.onvif_meta = {
+            "asset_type": "CAMERA",
+            "local_ip_address": "192.168.1.64",
+            "camera_access_key": "remote_user:2jCkrCRSeahzKEU:d5694af2-21e2-4a39-9bad-2fb98d9818bd",
+            "middleware_hostname": "dev_middleware.coronasafe.live",
+        }
+        cls.hl7monitor_meta = {}
+        cls.ventilator_meta = {}
+        cls.bed = cls.create_bed(
+            facility=cls.facility, location=cls.asset1_location, meta={}
+        )
+        cls.asset = cls.create_asset(location=cls.asset1_location)
+
+    def test_onvif_relative_move(self):
+        self.asset.asset_class = "ONVIF"
+        self.asset.meta = self.onvif_meta
+        self.asset.save()
+        boundary_asset_bed = AssetBed.objects.create(
+            asset=self.asset,
+            bed=self.bed,
+            meta={
+                "range": {
+                    "max_x": 2,
+                    "min_x": -2,
+                    "max_y": 2,
+                    "min_y": -2,
+                }
+            },
+        )
+        sample_data = {
+            "action": {
+                "type": "relative_move",
+                "data": {
+                    "x": 0.1,
+                    "y": 0.1,
+                    "zoom": 0.1,
+                    "camera_state": {"x": 1.5, "y": 1.5, "zoom": 0},
+                    "id": boundary_asset_bed.external_id,
+                },
+            }
+        }
+        response = self.client.post(
+            f"/api/v1/asset/{self.asset.external_id}/operate_assets/",
+            sample_data,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        sample_data_invald = {
+            "action": {
+                "type": "relative_move",
+                "data": {
+                    "x": 0.6,
+                    "y": 0.1,
+                    "zoom": 0.1,
+                    "camera_state": {"x": 1.5, "y": 1.5, "zoom": 0},
+                    "id": boundary_asset_bed.external_id,
+                },
+            }
+        }
+        response_invalid = self.client.post(
+            f"/api/v1/asset/{self.asset.external_id}/operate_assets/",
+            sample_data_invald,
+            "json",
+        )
+
+        self.assertEqual(response_invalid.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response_invalid.data.get("message", {}).get("action", {}).code, "invalid"
+        )
+
+    def test_hl7monitor(self):
+        self.asset.asset_class = "HL7MONITOR"
+        self.asset.meta = self.hl7monitor_meta
+        self.asset.save()
+        pass
+
+    def test_ventilator(self):
+        self.asset.asset_class = "VENTILATOR"
+        self.asset.meta = self.ventilator_meta
+        self.asset.save()
+        pass

care/facility/tests/test_patient_consultationbed.py

@@ -0,0 +1,105 @@
+import datetime
+
+from django.utils.timezone import make_aware
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from care.users.models import User
+from care.utils.tests.test_utils import TestUtils
+
+
+class TestPatientConsultationbed(TestUtils, APITestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.state = cls.create_state()
+        cls.district = cls.create_district(state=cls.state)
+        cls.local_body = cls.create_local_body(cls.district)
+        cls.user = cls.create_user(district=cls.district, username="test user")
+        cls.facility = cls.create_facility(
+            district=cls.district, local_body=cls.local_body, user=cls.user
+        )
+        cls.patient = cls.create_patient(cls.district, cls.facility)
+        cls.location = cls.create_asset_location(facility=cls.facility)
+        cls.bed = cls.create_bed(
+            name="Test Bed",
+            facility=cls.facility,
+            location=cls.location,
+        )
+        cls.consultation = cls.create_consultation(
+            facility=cls.facility, patient=cls.patient
+        )
+
+    def test_patient_privacy_toggle_success(self):
+        allowed_user_types = [
+            "DistrictAdmin",
+            "WardAdmin",
+            "LocalBodyAdmin",
+            "StateAdmin",
+            "Doctor",
+            "Staff",
+        ]
+        for user_type in allowed_user_types:
+            self.user = self.create_user(
+                username=f"{user_type} test user",
+                user_type=User.TYPE_VALUE_MAP[user_type],
+                district=self.district,
+                home_facility=self.facility,
+            )
+            consultation_bed = self.create_consultation_bed(
+                consultation=self.consultation,
+                bed=self.bed,
+                start_date=make_aware(datetime.datetime.now()),
+                end_date=make_aware(datetime.datetime.now()),
+                privacy=True,
+            )
+
+            self.client.force_authenticate(user=self.user)
+            response = self.client.patch(
+                f"/api/v1/consultationbed/{consultation_bed.external_id}/toggle_patient_privacy/",
+                {"external_id": consultation_bed.external_id},
+            )
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            consultation_bed.delete()
+            self.user.delete()
+
+    def test_patient_privacy_toggle_failure(self):
+        non_allowed_user_types = [
+            "Transportation",
+            "Pharmacist",
+            "Volunteer",
+            "StaffReadOnly",
+            "Reserved",
+            "DistrictLabAdmin",
+            "DistrictReadOnlyAdmin",
+            "StateLabAdmin",
+            "StateReadOnlyAdmin",
+            "Doctor",
+            "Staff",
+        ]
+        for user_type in non_allowed_user_types:
+            facility2 = self.create_facility(
+                district=self.district, local_body=self.local_body, user=self.user
+            )
+            self.user = self.create_user(
+                username=f"{user_type} test user",
+                user_type=User.TYPE_VALUE_MAP[user_type],
+                district=self.district,
+                home_facility=facility2,
+            )
+
+            consultation_bed = self.create_consultation_bed(
+                consultation=self.consultation,
+                bed=self.bed,
+                start_date=make_aware(datetime.datetime.now()),
+                end_date=make_aware(datetime.datetime.now()),
+                privacy=True,
+            )
+
+            self.client.force_authenticate(user=self.user)
+            response = self.client.patch(
+                f"/api/v1/consultationbed/{consultation_bed.external_id}/toggle_patient_privacy/",
+                {"external_id": consultation_bed.external_id},
+            )
+            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+            consultation_bed.delete()
+            self.user.delete()

care/utils/assetintegration/base.py

@@ -52,3 +52,6 @@
             return response
         except json.decoder.JSONDecodeError:
             return {"error": "Invalid Response"}
+
+    def validate_action(self, action):
+        pass

care/utils/assetintegration/onvif.py

@@ -29,7 +29,10 @@
     def handle_action(self, action):
         action_type = action["type"]
         action_data = action.get("data", {})
-
+        allowed_action_data = ["x", "y", "zoom"]
+        action_data = {
+            key: action_data[key] for key in action_data if key in allowed_action_data
+        }
         request_body = {
             "hostname": self.host,
             "port": 80,
@@ -55,3 +58,41 @@
             return self.api_post(self.get_url("relativeMove"), request_body)
 
         raise ValidationError({"action": "invalid action type"})
+
+    def validate_action(self, action):
+        from care.facility.models.bed import AssetBed
+
+        action_type = action["type"]
+        action_data = action.get("data", {})
+        boundary_preset_id = action_data.get("id", None)
+
+        if (
+            not boundary_preset_id
+            or action_type != self.OnvifActions.RELATIVE_MOVE.value
+        ):
+            return
+
+        boundary_preset = AssetBed.objects.filter(
+            external_id=boundary_preset_id
+        ).first()
+
+        if (
+            not boundary_preset
+            or not action_data.get("camera_state", None)
+            or not action_data["camera_state"].get("x", None)
+            or not action_data["camera_state"].get("y", None)
+        ):
+            raise ValidationError({"action": "invalid action type"})
+
+        boundary_range = boundary_preset.meta.get("range", None)
+        camera_state = action_data["camera_state"]
+
+        if (
+            (camera_state["x"] + action_data["x"] < boundary_range["min_x"])
+            or (camera_state["x"] + action_data["x"] > boundary_range["max_x"])
+            or (camera_state["y"] + action_data["y"] < boundary_range["min_y"])
+            or (camera_state["y"] + action_data["y"] > boundary_range["max_y"])
+        ):
+            raise ValidationError({"action": "invalid action type"})
+
+        return