sajad-dev/Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UPDATE

Browse Source
This commit is contained in:
Mohammad Sajad Pourajam
2026-03-26 15:39:31 +03:30
parent f305e00cfe
commit 32a0e3f3d9
26 changed files with 2188 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sensor_hub/management/__init__.py
+1
View File
@@ -0,0 +1 @@
sensor_hub/management/commands/__init__.py
+1
View File
@@ -0,0 +1 @@
sensor_hub/management/commands/seed_admin_sensor.py
+21
View File
@@ -0,0 +1,21 @@
from django.core.management.base import BaseCommand, CommandError
from sensor_hub.seeds import seed_admin_sensor
class Command(BaseCommand):
help = "Create or update the default full sensor for the admin user."
def handle(self, *args, **options):
try:
sensor, created = seed_admin_sensor()
except ValueError as exc:
raise CommandError(str(exc)) from exc
action = "created" if created else "updated"
self.stdout.write(
self.style.SUCCESS(
f"Admin sensor {action}: uuid_sensor={sensor.uuid_sensor}, "
f"name={sensor.name}, owner={sensor.owner.username}"
)
)
sensor_hub/seeds.py
+92
View File
@@ -0,0 +1,92 @@
import uuid
from django.db import transaction
from account.seeds import seed_admin_user
from .models import Sensor
ADMIN_SENSOR_UUID = uuid.UUID("11111111-1111-1111-1111-111111111111")
ADMIN_SENSOR_DATA = {
"name": "Admin Smart Farm Sensor",
"is_active": True,
"specifications": {
"model": "CL-SENSE-PRO-X",
"firmware": "2.4.1",
"manufacturer": "CropLogic",
"serial_number": "CL-ADMIN-0001",
"capabilities": [
"temperature",
"humidity",
"soil_moisture",
"soil_temperature",
"light_intensity",
"ph",
"ec",
"wind_speed",
],
"connectivity": {
"protocol": "LoRaWAN",
"sim_enabled": True,
"bluetooth": True,
"wifi_fallback": True,
},
"location": {
"label": "Admin Demo Field",
"lat": 35.6892,
"lng": 51.389,
"altitude_m": 1190,
},
},
"power_source": {
"type": "hybrid",
"battery": {
"capacity_mah": 12000,
"voltage": 12,
"health_percent": 98,
},
"solar": {
"panel_watt": 40,
"controller": "MPPT",
},
"backup": "dc_adapter",
},
"customized_sensors": {
"thresholds": {
"temperature_c": {"min": 10, "max": 36},
"humidity_percent": {"min": 30, "max": 85},
"soil_moisture_percent": {"min": 25, "max": 70},
"ph": {"min": 5.8, "max": 7.2},
"ec_ds_m": {"min": 1.1, "max": 2.4},
},
"report_interval_sec": 300,
"alerts": {
"sms": True,
"email": True,
"push": True,
},
"calibration": {
"last_calibrated_at": "2025-03-01T08:30:00Z",
"technician": "system",
"status": "passed",
},
},
}
@transaction.atomic
def seed_admin_sensor():
owner, _ = seed_admin_user()
sensor, created = Sensor.objects.update_or_create(
uuid_sensor=ADMIN_SENSOR_UUID,
defaults={
"owner": owner,
"name": ADMIN_SENSOR_DATA["name"],
"is_active": ADMIN_SENSOR_DATA["is_active"],
"specifications": ADMIN_SENSOR_DATA["specifications"],
"power_source": ADMIN_SENSOR_DATA["power_source"],
"customized_sensors": ADMIN_SENSOR_DATA["customized_sensors"],
},
)
return sensor, created
sensor_hub/serializers.py
+4
View File
@@ -30,3 +30,7 @@ class SensorCreateSerializer(serializers.ModelSerializer):
"power_source",
"customized_sensors",
]
class SensorToggleSerializer(serializers.Serializer):
uuid_sensor = serializers.UUIDField()
sensor_hub/urls.py
+5 -5
View File
@@ -1,10 +1,10 @@
from django.urls import path
from .views import SensorHubView
from .views import SensorActiveView, SensorDeactiveView, SensorDetailView, SensorListCreateView
urlpatterns = [
path("active/", SensorHubView.as_view(), name="sensor-hub-active", kwargs={"action": "active"}),
path("deactive/", SensorHubView.as_view(), name="sensor-hub-deactive", kwargs={"action": "deactive"}),
path("<uuid:uuid>/", SensorHubView.as_view(), name="sensor-hub-detail"),
path("", SensorHubView.as_view(), name="sensor-hub-list"),
path("active/", SensorActiveView.as_view(), name="sensor-hub-active"),
path("deactive/", SensorDeactiveView.as_view(), name="sensor-hub-deactive"),
path("<uuid:uuid>/", SensorDetailView.as_view(), name="sensor-hub-detail"),
path("", SensorListCreateView.as_view(), name="sensor-hub-list"),
]
sensor_hub/views.py
+81 -98
View File
@@ -1,64 +1,15 @@
from rest_framework import status
from rest_framework import serializers
from rest_framework import serializers, status
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import extend_schema, extend_schema_view
from drf_spectacular.utils import extend_schema
from config.swagger import code_response
from .models import Sensor
from .serializers import SensorCreateSerializer, SensorSerializer
from .serializers import SensorCreateSerializer, SensorSerializer, SensorToggleSerializer
@extend_schema_view(
get=extend_schema(
tags=["Sensor Hub"],
responses={
200: code_response("SensorHubGetResponse", data=serializers.JSONField()),
404: code_response("SensorHubNotFoundResponse"),
},
),
post=extend_schema(
tags=["Sensor Hub"],
request=OpenApiTypes.OBJECT,
responses={
201: code_response("SensorCreateResponse", data=serializers.JSONField()),
200: code_response("SensorToggleResponse"),
400: code_response("SensorToggleValidationResponse"),
404: code_response("SensorToggleNotFoundResponse"),
},
),
patch=extend_schema(
tags=["Sensor Hub"],
request=SensorCreateSerializer,
responses={
200: code_response("SensorUpdateResponse", data=SensorSerializer()),
404: code_response("SensorUpdateNotFoundResponse"),
},
),
delete=extend_schema(
tags=["Sensor Hub"],
responses={
200: code_response("SensorDeleteResponse"),
404: code_response("SensorDeleteNotFoundResponse"),
},
),
)
class SensorHubView(APIView):
"""
Sensor-hub CRUD endpoints connected to the database.
Routes:
- GET "" → List sensors for authenticated user.
- GET "<uuid>/" → Detail of a single sensor.
- POST "" → Create a new sensor.
- PATCH "<uuid>/" → Update an existing sensor.
- DELETE "<uuid>/" → Delete a sensor.
- POST "active/" → Activate a sensor (requires uuid_sensor in body).
- POST "deactive/" → Deactivate a sensor (requires uuid_sensor in body).
"""
class SensorHubBaseView(APIView):
permission_classes = [IsAuthenticated]
def _get_sensor(self, request, uuid):
@@ -67,74 +18,106 @@ class SensorHubView(APIView):
except Sensor.DoesNotExist:
return None
def get(self, request, *args, **kwargs):
uuid = kwargs.get("uuid")
if uuid is not None:
sensor = self._get_sensor(request, uuid)
if sensor is None:
return Response(
{"code": 404, "msg": "Sensor not found."},
status=status.HTTP_404_NOT_FOUND,
)
data = SensorSerializer(sensor).data
return Response({"code": 200, "msg": "success", "data": data}, status=status.HTTP_200_OK)
class SensorListCreateView(SensorHubBaseView):
@extend_schema(
tags=["Sensor Hub"],
responses={200: code_response("SensorListResponse", data=SensorSerializer(many=True))},
)
def get(self, request):
sensors = Sensor.objects.filter(owner=request.user)
data = SensorSerializer(sensors, many=True).data
return Response({"code": 200, "msg": "success", "data": data}, status=status.HTTP_200_OK)
def post(self, request, *args, **kwargs):
action = kwargs.get("action")
if action in ("active", "deactive"):
return self._toggle_active(request, is_active=(action == "active"))
@extend_schema(
tags=["Sensor Hub"],
request=SensorCreateSerializer,
responses={201: code_response("SensorCreateResponse", data=SensorSerializer())},
)
def post(self, request):
serializer = SensorCreateSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
sensor = serializer.save(owner=request.user)
data = SensorSerializer(sensor).data
return Response(
{"code": 201, "msg": "success", "data": data},
status=status.HTTP_201_CREATED,
)
return Response({"code": 201, "msg": "success", "data": data}, status=status.HTTP_201_CREATED)
def patch(self, request, *args, **kwargs):
uuid = kwargs.get("uuid")
class SensorDetailView(SensorHubBaseView):
@extend_schema(
tags=["Sensor Hub"],
responses={
200: code_response("SensorDetailResponse", data=SensorSerializer()),
404: code_response("SensorNotFoundResponse"),
},
)
def get(self, request, uuid):
sensor = self._get_sensor(request, uuid)
if sensor is None:
return Response(
{"code": 404, "msg": "Sensor not found."},
status=status.HTTP_404_NOT_FOUND,
)
return Response({"code": 404, "msg": "Sensor not found."}, status=status.HTTP_404_NOT_FOUND)
data = SensorSerializer(sensor).data
return Response({"code": 200, "msg": "success", "data": data}, status=status.HTTP_200_OK)
@extend_schema(
tags=["Sensor Hub"],
request=SensorCreateSerializer,
responses={
200: code_response("SensorUpdateResponse", data=SensorSerializer()),
404: code_response("SensorUpdateNotFoundResponse"),
},
)
def patch(self, request, uuid):
sensor = self._get_sensor(request, uuid)
if sensor is None:
return Response({"code": 404, "msg": "Sensor not found."}, status=status.HTTP_404_NOT_FOUND)
serializer = SensorCreateSerializer(sensor, data=request.data, partial=True)
serializer.is_valid(raise_exception=True)
serializer.save()
data = SensorSerializer(sensor).data
return Response({"code": 200, "msg": "success", "data": data}, status=status.HTTP_200_OK)
def delete(self, request, *args, **kwargs):
uuid = kwargs.get("uuid")
@extend_schema(
tags=["Sensor Hub"],
responses={
200: code_response("SensorDeleteResponse"),
404: code_response("SensorDeleteNotFoundResponse"),
},
)
def delete(self, request, uuid):
sensor = self._get_sensor(request, uuid)
if sensor is None:
return Response(
{"code": 404, "msg": "Sensor not found."},
status=status.HTTP_404_NOT_FOUND,
)
return Response({"code": 404, "msg": "Sensor not found."}, status=status.HTTP_404_NOT_FOUND)
sensor.delete()
return Response({"code": 200, "msg": "success"}, status=status.HTTP_200_OK)
def _toggle_active(self, request, is_active):
uuid_sensor = request.data.get("uuid_sensor")
if not uuid_sensor:
return Response(
{"code": 400, "msg": "uuid_sensor is required."},
status=status.HTTP_400_BAD_REQUEST,
)
sensor = self._get_sensor(request, uuid_sensor)
class SensorToggleView(SensorHubBaseView):
action = None
@extend_schema(
tags=["Sensor Hub"],
request=SensorToggleSerializer,
responses={
200: code_response("SensorToggleResponse"),
400: code_response("SensorToggleValidationResponse"),
404: code_response("SensorToggleNotFoundResponse"),
},
)
def post(self, request):
serializer = SensorToggleSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
sensor = self._get_sensor(request, serializer.validated_data["uuid_sensor"])
if sensor is None:
return Response(
{"code": 404, "msg": "Sensor not found."},
status=status.HTTP_404_NOT_FOUND,
)
sensor.is_active = is_active
return Response({"code": 404, "msg": "Sensor not found."}, status=status.HTTP_404_NOT_FOUND)
sensor.is_active = self.action == "active"
sensor.save(update_fields=["is_active", "updated_at"])
return Response({"code": 200, "msg": "success"}, status=status.HTTP_200_OK)
class SensorActiveView(SensorToggleView):
action = "active"
class SensorDeactiveView(SensorToggleView):
action = "deactive"