UPDATE
This commit is contained in:
@@ -0,0 +1,7 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class SensorCatalogConfig(AppConfig):
|
||||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "sensor_catalog"
|
||||
verbose_name = "Sensor Catalog"
|
||||
@@ -0,0 +1,57 @@
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from sensor_catalog.models import SensorCatalog
|
||||
|
||||
|
||||
SENSOR_CATALOG_ITEMS = [
|
||||
{
|
||||
"name": "Sensor 7 - Soil Moisture Sensor v1.2",
|
||||
"description": (
|
||||
"This sensor is typically the YL-69 or FC-28 soil moisture sensor. "
|
||||
"It measures only soil moisture and provides analog and digital outputs. "
|
||||
"It does not report soil temperature, pH, or nutrients."
|
||||
),
|
||||
"customizable_fields": [],
|
||||
"supported_power_sources": ["solar", "direct_power"],
|
||||
"returned_data_fields": ["soil_moisture", "analog_output", "digital_output"],
|
||||
"sample_payload": {
|
||||
"soil_moisture": 42,
|
||||
"analog_output": 610,
|
||||
"digital_output": 1,
|
||||
},
|
||||
"is_active": True,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "Seed sensor catalog data."
|
||||
|
||||
def handle(self, *args, **options):
|
||||
created_count = 0
|
||||
updated_count = 0
|
||||
|
||||
for item in SENSOR_CATALOG_ITEMS:
|
||||
sensor, created = SensorCatalog.objects.update_or_create(
|
||||
name=item["name"],
|
||||
defaults={
|
||||
"description": item["description"],
|
||||
"customizable_fields": item["customizable_fields"],
|
||||
"supported_power_sources": item["supported_power_sources"],
|
||||
"returned_data_fields": item["returned_data_fields"],
|
||||
"sample_payload": item["sample_payload"],
|
||||
"is_active": item["is_active"],
|
||||
},
|
||||
)
|
||||
if created:
|
||||
created_count += 1
|
||||
self.stdout.write(self.style.SUCCESS(f"Created sensor catalog item: {sensor.name}"))
|
||||
else:
|
||||
updated_count += 1
|
||||
self.stdout.write(self.style.WARNING(f"Updated sensor catalog item: {sensor.name}"))
|
||||
|
||||
self.stdout.write(
|
||||
self.style.SUCCESS(
|
||||
f"Sensor catalog seeding complete. Created: {created_count}, Updated: {updated_count}"
|
||||
)
|
||||
)
|
||||
@@ -0,0 +1,32 @@
|
||||
# Generated by Django 5.2.12 on 2026-03-20 00:00
|
||||
|
||||
import uuid
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
initial = True
|
||||
|
||||
dependencies = []
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name="SensorCatalog",
|
||||
fields=[
|
||||
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
|
||||
("uuid", models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, unique=True)),
|
||||
("name", models.CharField(db_index=True, max_length=255, unique=True)),
|
||||
("description", models.TextField(blank=True, default="")),
|
||||
("customizable_fields", models.JSONField(blank=True, default=list)),
|
||||
("returned_data_fields", models.JSONField(blank=True, default=list)),
|
||||
("sample_payload", models.JSONField(blank=True, default=dict)),
|
||||
("is_active", models.BooleanField(default=True)),
|
||||
("created_at", models.DateTimeField(auto_now_add=True)),
|
||||
("updated_at", models.DateTimeField(auto_now=True)),
|
||||
],
|
||||
options={
|
||||
"db_table": "sensor_catalogs",
|
||||
"ordering": ["name"],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,17 @@
|
||||
# Generated by Django 5.2.12 on 2026-03-20 01:00
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
dependencies = [
|
||||
("sensor_catalog", "0001_initial"),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name="sensorcatalog",
|
||||
name="supported_power_sources",
|
||||
field=models.JSONField(blank=True, default=list),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,23 @@
|
||||
import uuid
|
||||
|
||||
from django.db import models
|
||||
|
||||
|
||||
class SensorCatalog(models.Model):
|
||||
uuid = models.UUIDField(default=uuid.uuid4, unique=True, editable=False, db_index=True)
|
||||
name = models.CharField(max_length=255, unique=True, db_index=True)
|
||||
description = models.TextField(blank=True, default="")
|
||||
customizable_fields = models.JSONField(default=list, blank=True)
|
||||
supported_power_sources = models.JSONField(default=list, blank=True)
|
||||
returned_data_fields = models.JSONField(default=list, blank=True)
|
||||
sample_payload = models.JSONField(default=dict, blank=True)
|
||||
is_active = models.BooleanField(default=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
class Meta:
|
||||
db_table = "sensor_catalogs"
|
||||
ordering = ["name"]
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
@@ -0,0 +1,19 @@
|
||||
from rest_framework import serializers
|
||||
|
||||
from .models import SensorCatalog
|
||||
|
||||
|
||||
class SensorCatalogSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = SensorCatalog
|
||||
fields = [
|
||||
"uuid",
|
||||
"name",
|
||||
"description",
|
||||
"customizable_fields",
|
||||
"supported_power_sources",
|
||||
"returned_data_fields",
|
||||
"sample_payload",
|
||||
"is_active",
|
||||
]
|
||||
read_only_fields = fields
|
||||
@@ -0,0 +1,55 @@
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
from rest_framework.test import APIRequestFactory, force_authenticate
|
||||
|
||||
from sensor_catalog.models import SensorCatalog
|
||||
from sensor_catalog.views import SensorCatalogListView
|
||||
|
||||
|
||||
class SensorCatalogListViewTests(TestCase):
|
||||
def setUp(self):
|
||||
self.factory = APIRequestFactory()
|
||||
self.user = get_user_model().objects.create_user(
|
||||
username="sensor-user",
|
||||
password="secret123",
|
||||
email="sensor@example.com",
|
||||
phone_number="09120000002",
|
||||
)
|
||||
SensorCatalog.objects.update_or_create(
|
||||
name="Sensor 7 - Soil Moisture Sensor v1.2",
|
||||
defaults={
|
||||
"description": (
|
||||
"Measures only soil moisture using electrical resistance between two metal probes. "
|
||||
"Provides analog and digital outputs."
|
||||
),
|
||||
"customizable_fields": [],
|
||||
"supported_power_sources": ["solar", "direct_power"],
|
||||
"returned_data_fields": ["soil_moisture", "analog_output", "digital_output"],
|
||||
"sample_payload": {"soil_moisture": 42, "analog_output": 610, "digital_output": 1},
|
||||
"is_active": True,
|
||||
},
|
||||
)
|
||||
SensorCatalog.objects.update_or_create(
|
||||
name="Legacy Sensor",
|
||||
defaults={
|
||||
"customizable_fields": [],
|
||||
"supported_power_sources": ["direct_power"],
|
||||
"returned_data_fields": ["status"],
|
||||
"sample_payload": {"status": "offline"},
|
||||
"is_active": False,
|
||||
},
|
||||
)
|
||||
|
||||
def test_list_returns_all_existing_sensors(self):
|
||||
request = self.factory.get("/api/sensor-catalog/")
|
||||
force_authenticate(request, user=self.user)
|
||||
|
||||
response = SensorCatalogListView.as_view()(request)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.data["code"], 200)
|
||||
self.assertEqual(len(response.data["data"]), 2)
|
||||
self.assertEqual(
|
||||
{item["name"] for item in response.data["data"]},
|
||||
{"Sensor 7 - Soil Moisture Sensor v1.2", "Legacy Sensor"},
|
||||
)
|
||||
@@ -0,0 +1,7 @@
|
||||
from django.urls import path
|
||||
|
||||
from .views import SensorCatalogListView
|
||||
|
||||
urlpatterns = [
|
||||
path("", SensorCatalogListView.as_view(), name="sensor-catalog-list"),
|
||||
]
|
||||
@@ -0,0 +1,22 @@
|
||||
from rest_framework import status
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.views import APIView
|
||||
from drf_spectacular.utils import extend_schema
|
||||
|
||||
from config.swagger import code_response
|
||||
from .models import SensorCatalog
|
||||
from .serializers import SensorCatalogSerializer
|
||||
|
||||
|
||||
class SensorCatalogListView(APIView):
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
@extend_schema(
|
||||
tags=["Sensor Catalog"],
|
||||
responses={200: code_response("SensorCatalogListResponse", data=SensorCatalogSerializer(many=True))},
|
||||
)
|
||||
def get(self, request):
|
||||
sensors = SensorCatalog.objects.order_by("name")
|
||||
data = SensorCatalogSerializer(sensors, many=True).data
|
||||
return Response({"code": 200, "msg": "success", "data": data}, status=status.HTTP_200_OK)
|
||||
Reference in New Issue
Block a user