Sync hardware-team with master

ag_data/serializers.py

@@ -0,0 +1,20 @@
+from rest_framework import serializers
+
+from ag_data.models import AGEvent, AGSensor, AGMeasurement
+
+
+class AGMeasurementSerializer(serializers.ModelSerializer):
+    """
+    Serializer for the Model AGMeasurement.
+    """
+
+    sensor_id = serializers.PrimaryKeyRelatedField(
+        read_only=False, queryset=AGSensor.objects.all()
+    )
+    event_uuid = serializers.PrimaryKeyRelatedField(
+        read_only=False, queryset=AGEvent.objects.all()
+    )
+
+    class Meta:
+        model = AGMeasurement
+        fields = ("timestamp", "sensor_id", "event_uuid", "value")

mercury/grafanaAPI/grafana_api.py

@@ -93,7 +93,7 @@ def get_dashboard_by_event_name(self, event_name):
         # If there are spaces in the name, the GF API will replace them with dashes
         # to generate the "slug". A slug can be used to query the API.
         endpoint = os.path.join(
-            self.hostname, "api/dashboards/db", event_name.lower().replace(" ", "-"),
+            self.hostname, "api/dashboards/db", event_name.lower().replace(" ", "-")
         )
         response = requests.get(url=endpoint, auth=("api_key", self.api_token))
 

mercury/tests/test_configure_sensors.py

@@ -34,10 +34,7 @@ class TestConfigureSensorView(TestCase):
     test_sensor_update_object_name = "update_sensor"
     test_type_update_object_name = "update_type"
 
-    test_sensor = {
-        "name": "wind speed sensor",
-        "type_id": test_type_object_name,
-    }
+    test_sensor = {"name": "wind speed sensor", "type_id": test_type_object_name}
 
     test_sensor_type = {
         "type-name": "fuel level",
@@ -57,14 +54,8 @@ def setUp(self):
             name=self.test_type_object_name,
             processing_formula=0,
             format={
-                self.field_name_1: {
-                    "data_type": self.data_type_1,
-                    "unit": self.unit_1,
-                },
-                self.field_name_2: {
-                    "data_type": self.data_type_1,
-                    "unit": self.unit_1,
-                },
+                self.field_name_1: {"data_type": self.data_type_1, "unit": self.unit_1},
+                self.field_name_2: {"data_type": self.data_type_1, "unit": self.unit_1},
             },
         )
         test_type_object.save()
@@ -78,14 +69,8 @@ def setUp(self):
             name=self.test_type_update_object_name,
             processing_formula=0,
             format={
-                self.field_name_1: {
-                    "data_type": self.data_type_1,
-                    "unit": self.unit_1,
-                },
-                self.field_name_2: {
-                    "data_type": self.data_type_1,
-                    "unit": self.unit_1,
-                },
+                self.field_name_1: {"data_type": self.data_type_1, "unit": self.unit_1},
+                self.field_name_2: {"data_type": self.data_type_1, "unit": self.unit_1},
             },
         )
 

mercury/tests/test_grafana.py

@@ -354,7 +354,7 @@ def test_create_postgres_datasource(self):
 
         # Query new datasource
         endpoint = os.path.join(
-            self.grafana.endpoints["datasource_name"], self.datasource_name,
+            self.grafana.endpoints["datasource_name"], self.datasource_name
         )
         headers = {"Content-Type": "application/json"}
         response = requests.get(
@@ -388,7 +388,7 @@ def test_delete_postgres_datasource(self):
 
         # confirm that the datasource was actually deleted by querying it
         endpoint = os.path.join(
-            self.grafana.endpoints["datasource_name"], self.datasource_name,
+            self.grafana.endpoints["datasource_name"], self.datasource_name
         )
         headers = {"Content-Type": "application/json"}
         response = requests.get(

mercury/tests/test_measurement.py

@@ -0,0 +1,85 @@
+import datetime
+
+import mock
+from django.test import TestCase
+from rest_framework.reverse import reverse
+
+from ag_data.models import AGEvent, AGVenue
+
+
+def fake_venue():
+    return AGVenue(
+        uuid="d81cac8d-26e1-4983-a942-1922e54a943a",
+        name="fake venue",
+        description="fake venue",
+        latitude=111.11,
+        longitude=111.11,
+    )
+
+
+def fake_event(uuid):
+    """
+        Mock a dummy AGEvent model
+    """
+    return AGEvent(
+        uuid=uuid,
+        name="fake event",
+        description="fake event",
+        date=datetime.datetime(2020, 2, 2, 20, 21, 22),
+        venue_uuid=fake_venue(),
+    )
+
+
+def fake_valid(res, raise_exception=True):
+    return True
+
+
+class TestMeasurement(TestCase):
+    def setUp(self) -> None:
+        self.post_url = "mercury:measurement"
+        self.uuid = "d81cac8d-26e1-4983-a942-1922e54a943d"
+        self.uuid2 = "d81cac8d-26e1-4983-a942-1922e54a943a"
+
+    def post_radio_data(self):
+        # POST sensor data to the measurement url
+        response = self.client.post(
+            reverse(self.post_url, args=[self.uuid]),
+            data={
+                "sensor_id": 1,
+                "values": {"power": "2", "speed": 1},
+                "date": datetime.datetime(2020, 2, 2, 20, 21, 22),
+            },
+        )
+        return response
+
+    def post_defect_data(self):
+        response = self.client.post(
+            reverse(self.post_url, args=[self.uuid]),
+            data={
+                "values": {"power": "2", "speed": 1},
+                "date": datetime.datetime(2020, 2, 2, 20, 21, 22),
+            },
+        )
+        return response
+
+    def test_Radio_Receiver_POST_Event_Not_Exist(self):
+        response = self.client.post(reverse(self.post_url, args=[self.uuid2]))
+        self.assertEqual(404, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    def test_Radio_Receiver_POST_Missing_Params(self):
+        response = self.post_defect_data()
+        self.assertEqual(400, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    def test_Radio_Receiver_POST_Fail_to_Save(self):
+        response = self.post_radio_data()
+        self.assertEqual(400, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    @mock.patch("ag_data.serializers.AGMeasurementSerializer.is_valid", fake_valid)
+    @mock.patch("ag_data.serializers.AGMeasurementSerializer.save", fake_valid)
+    @mock.patch("ag_data.serializers.AGMeasurementSerializer.data", "")
+    def test_Radio_Receiver_POST_Event_Success(self):
+        response = self.post_radio_data()
+        self.assertEqual(201, response.status_code)

mercury/tests/test_radio_receiver.py

@@ -0,0 +1,132 @@
+from django.test import TestCase
+from django.urls import reverse
+import datetime
+import mock
+
+from ag_data.models import AGEvent, AGVenue
+
+
+def fake_venue():
+    return AGVenue(
+        uuid="d81cac8d-26e1-4983-a942-1922e54a943a",
+        name="fake venue",
+        description="fake venue",
+        latitude=111.11,
+        longitude=111.11,
+    )
+
+
+def fake_event(uuid):
+    """
+        Mock a dummy AGEvent model
+    """
+    return AGEvent(
+        uuid=uuid,
+        name="fake event",
+        description="fake event",
+        date=datetime.datetime(2020, 2, 2, 20, 21, 22),
+        venue_uuid=fake_venue(),
+    )
+
+
+def fake_valid(res):
+    return True
+
+
+def fake_valid_port():
+    return ["dev/tty.USB"]
+
+
+def fake_invalid_port():
+    return []
+
+
+class TestRadioReceiverView(TestCase):
+    def setUp(self) -> None:
+        self.get_url = "mercury:radioreceiver"
+        self.post_url = "mercury:radioreceiver"
+        self.uuid = "d81cac8d-26e1-4983-a942-1922e54a943d"
+        self.uuid2 = "d81cac8d-26e1-4983-a942-1922e54a943a"
+
+    def test_Radio_Receiver_GET_No_Related_Event(self):
+        response = self.client.get(reverse(self.get_url, args=[self.uuid2]))
+        self.assertEqual(404, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    def test_Radio_Receiver_GET_Missing_Enable(self):
+        response = self.client.get(
+            reverse(self.get_url, args=[self.uuid]),
+            data={
+                "baudrate": 9000,
+                "bytesize": 8,
+                "parity": "N",
+                "stopbits": 1,
+                "timeout": 1,
+            },
+        )
+        self.assertEqual(400, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    @mock.patch("mercury.views.radioreceiver.serial_ports", fake_invalid_port)
+    def test_Radio_Receiver_GET_No_Valid_Port(self):
+        response = self.client.get(
+            reverse(self.get_url, args=[self.uuid]),
+            data={
+                "enable": 1,
+                "baudrate": 9000,
+                "bytesize": 8,
+                "parity": "N",
+                "stopbits": 1,
+                "timeout": 1,
+            },
+        )
+        self.assertEqual(503, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    @mock.patch("mercury.views.radioreceiver.serial_ports", fake_valid_port)
+    def test_Radio_Receiver_GET_Success(self):
+        response = self.client.get(
+            reverse(self.get_url, args=[self.uuid]),
+            data={
+                "enable": 1,
+                "baudrate": 9000,
+                "bytesize": 8,
+                "parity": "N",
+                "stopbits": 1,
+                "timeout": 1,
+            },
+        )
+        self.assertEqual(200, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    @mock.patch("mercury.views.radioreceiver.serial_ports", fake_valid_port)
+    @mock.patch("mercury.views.radioreceiver.check_port", fake_valid)
+    def test_Radio_Receiver_GET_Close_Port_Success(self):
+        response = self.client.get(
+            reverse(self.get_url, args=[self.uuid]),
+            data={
+                "enable": 0,
+                "baudrate": 9000,
+                "bytesize": 8,
+                "parity": "N",
+                "stopbits": 1,
+                "timeout": 1,
+            },
+        )
+        self.assertEqual(200, response.status_code)
+
+    @mock.patch("ag_data.models.AGEvent.objects.get", fake_event)
+    def test_Radio_Receiver_Fake_GET_Success(self):
+        response = self.client.get(
+            reverse(self.get_url, args=[self.uuid]),
+            data={
+                "enable": 1,
+                "baudrate": 9000,
+                "bytesize": 8,
+                "parity": "N",
+                "stopbits": 1,
+                "timeout": 1,
+                "fake": 1,
+            },
+        )
+        self.assertEqual(200, response.status_code)

mercury/urls.py

@@ -4,7 +4,9 @@
     sensor,
     events,
     pitcrew,
+    radioreceiver,
     gf_config,
+    measurement,
 )
 
 app_name = "mercury"
@@ -40,11 +42,21 @@
     path("events/export/<uuid:event_uuid>/csv", events.export_event),
     path("events/export/<uuid:event_uuid>/json", events.export_event),
     path("pitcrew/", pitcrew.PitCrewView.as_view(), name="pitcrew"),
+    path(
+        "radioreceiver/<uuid:event_uuid>",
+        radioreceiver.RadioReceiverView.as_view(),
+        name="radioreceiver",
+    ),
     path("gfconfig/", gf_config.GFConfigView.as_view(), name="gfconfig"),
     path(
         "gfconfig/delete/<int:gf_id>", gf_config.delete_config, name="gfconfig_delete"
     ),
     path(
         "gfconfig/update/<int:gf_id>", gf_config.update_config, name="gfconfig_update"
     ),
+    path(
+        "measurement/<uuid:event_uuid>",
+        measurement.MeasurementView.as_view(),
+        name="measurement",
+    ),
 ]

mercury/views/events.py

@@ -77,10 +77,7 @@ def export_event(request, event_uuid=None, file_format="CSV"):
                     }
                     measurement_info.append(temp)
 
-            data = {
-                "event_info": event_info,
-                "measurement_info": measurement_info,
-            }
+            data = {"event_info": event_info, "measurement_info": measurement_info}
 
             response = HttpResponse(str(data), content_type="application/json")
             response["Content-Disposition"] = (