尚虎雲產銷平台-生產端軟體新增刪除感測器的功能

筆者撰寫此文時，尚虎雲團隊已新增刪除感測器的功能，因此本文和放在Github上開源的程式碼略有不同。

已更新的Github程式：https://github.com/shanghuyun/Shanghuyun-Production-side

本文列出的程式碼是上一版本，有關新增刪除感測器的功能會用不同顏色加以區分。

• urls.py

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

# -*- encoding: utf-8 -*- """ Copyright (c) 2019 - present AppSeed.us """ from django.urls import path, re_path from apps.home import views urlpatterns = [ # The home page path('', views.index, name='home'), path('add_sensor/<str:sensor_name>/', views.add_sensor_data, name='add_sensor_data'), path('delete_sensor/<str:sensor_name>/', views.delete_sensor, name='delete_sensor'), path('delete_product/<int:product_id>/', views.delete_product, name='delete_product'), path('sensor_data_list/<str:sensor_name>/', views.sensor_data_list, name='sensor_data_list'), # 獲取感測器數據 path('get_sensors/', views.get_sensors, name='get_sensors'), path('get_sensor_data/<str:sensor_name>/', views.get_sensor_data, name='get_sensor_data'), #API介面 path('api/get_counts/', views.get_counts, name='get_counts'), path('api/get_sensor_names/', views.get_sensor_names, name='get_sensor_names'), path('api/get_sensor_data_by_id/<str:sensor_name>/<int:count>/', views.get_sensor_data_by_name, name='get_sensor_data_by_name'), path('api/get_product_ids/', views.get_product_ids, name='get_product_ids'), path('api/get_product_details/<str:product_id>/', views.get_product_details, name='get_product_details'), # Matches any html file re_path(r'^.*\.html$', views.pages, name='pages'), ]

• views.py

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354

from django import template from django.contrib.auth.decorators import login_required from django.http import HttpResponse, HttpResponseRedirect, JsonResponse from django.core.paginator import Paginator from django.template import loader from django.urls import reverse from django.shortcuts import render, get_object_or_404 from .models import ( Color, Profile, Product, ProductImage, Sensor, SensorData, ) from .forms import ( ColorForm, ProductForm, ProductImageForm, ProfileForm, SensorForm, ) import traceback from datetime import datetime, timedelta def handle_color_form(request): if request.method == 'POST' and "color" in request.POST: color_form = ColorForm(request.POST) if color_form.is_valid(): color = color_form.cleaned_data['color'] gradient = color_form.cleaned_data['gradient'] gradient_color = color_form.cleaned_data['gradient_color'] Color.objects.update_or_create( defaults={'color': color, 'gradient': gradient, 'gradient_color': gradient_color} ) else: try: color_instance = Color.objects.latest('id') initial_data = { 'color': color_instance.color, 'gradient': color_instance.gradient, 'gradient_color': color_instance.gradient_color, } except Color.DoesNotExist: color_instance = Color.objects.create() initial_data = { 'color': color_instance.color, 'gradient': color_instance.gradient, 'gradient_color': color_instance.gradient_color, } color_form = ColorForm(initial=initial_data) return color_form def handle_background_color(): background_color, _ = Color.objects.get_or_create() return background_color @login_required(login_url="/login/") def index(request): color_form = handle_color_form(request) profile, created = Profile.objects.get_or_create(user=request.user) context = { 'segment': 'index', 'profile': profile, 'background_color': handle_background_color(), 'color_form': color_form, } html_template = loader.get_template('home/index.html') return HttpResponse(html_template.render(context, request)) @login_required(login_url="/login/") def pages(request): context = {} try: color_form = handle_color_form(request) profile, created = Profile.objects.get_or_create(user=request.user) context.update({ "profile": profile, "email": request.user.email, "background_color": handle_background_color(), "color_form": color_form, }) load_template = request.path.split('/')[-1] # if load_template == 'admin': # return HttpResponseRedirect(reverse('admin:index')) if load_template == "document.html": context["responses"] = [ {"status_code": 200, "status": "OK", "context": "Data added successfully", "solution": "代表資料成功傳送"}, {"status_code": 400, "status": "Bad Request", "context": "Sensor name is required", "solution": "未提供感測器名稱"}, {"status_code": 400, "status": "Bad Request", "context": "Data must be a valid number", "solution": "數據必須為整數或浮點數"}, {"status_code": 400, "status": "Bad Request", "context": "Data is required", "solution": "未供數據資料"}, {"status_code": 404, "status": "Not Found", "context": "Sensor not found", "solution": "請確認感測器名稱是否正確或是該感測器未建立"}, {"status_code": 500, "status": "Internal Server Error", "context": "An error occurred", "solution": "伺服器內部出現問題"} ] elif load_template == 'user.html': if request.method == "POST" and "color" not in request.POST: form = ProfileForm(request.POST, request.FILES, instance=profile) if form.is_valid(): form.save() return HttpResponseRedirect(request.path + '?save_success=True') else: form = ProfileForm(instance=profile) context.update({ 'form': form, 'username': request.user.username, }) elif load_template == 'add_product.html': if request.method == 'POST' and "color" not in request.POST: form = ProductForm(request.POST, request.FILES, profile_real_name=profile.real_name) image_form = ProductImageForm(request.POST, request.FILES) if form.is_valid() and image_form.is_valid(): price = form.cleaned_data.get('price') if price < 0: form.add_error('price', '價錢不得小於0') else: product = form.save(commit=False) product.seller = request.user product.save() images = request.FILES.getlist('multiImageInput') for image in images: ProductImage.objects.create(product=product, image=image) return HttpResponseRedirect(request.path + '?save_success=True&') else: print(form.errors, image_form.errors) else: form = ProductForm(initial={ 'email': request.user.email, 'phone_number': profile.phone_number, 'address': profile.contact_address }, profile_real_name=profile.real_name) image_form = ProductImageForm() if not profile.real_name or not profile.phone_number or not profile.contact_address: context['notification'] = True context.update({ 'form': form, 'image_form': image_form, }) elif load_template == 'product_list.html': context['products'] = Product.objects.order_by('name') elif load_template == 'add_sensor.html': if request.method == 'POST' and "color" not in request.POST: form = SensorForm(request.POST) if form.is_valid(): # 檢查是否有相同名稱的感測器 sensor_name = form.cleaned_data['name'] if Sensor.objects.filter(name=sensor_name).exists(): form.add_error('name', '具有相同名稱的感測器已存在') context['has_errors'] = form.errors else: form.save() context['sensor_add_success'] = True else: form = SensorForm() context['form'] = form context['has_errors'] = form.errors elif load_template == "sensor_list.html": sensors = Sensor.objects.all() sensor_list = [{ 'name': sensor.name, 'timestamp': SensorData.objects.filter(name=sensor).order_by('-timestamp').first().timestamp if SensorData.objects.filter(name=sensor).exists() else 'n/a', 'data': SensorData.objects.filter(name=sensor).order_by('-timestamp').first().data if SensorData.objects.filter(name=sensor).exists() else 'n/a', } for sensor in sensors] context['sensors'] = sensor_list context['segment'] = load_template html_template = loader.get_template('home/' + load_template) return HttpResponse(html_template.render(context, request)) except template.TemplateDoesNotExist: traceback.print_exc() html_template = loader.get_template('home/page-404.html') return HttpResponse(html_template.render(context, request)) except Exception: traceback.print_exc() html_template = loader.get_template('home/page-500.html') return HttpResponse(html_template.render(context, request)) @login_required(login_url="/login/") def delete_product(request, product_id): try: color_form = handle_color_form(request) background_color = handle_background_color() profile, created = Profile.objects.get_or_create(user=request.user) context = { "background_color": background_color, "color_form" : color_form, "profile" : profile } product = get_object_or_404(Product, id=product_id) if request.method == "POST": product.delete() context["delete"] = True context["products"] = Product.objects.all() html_template = loader.get_template('home/product_list.html') return HttpResponse(html_template.render(context, request)) else: context["delete"] = False context["products"] = Product.objects.all() html_template = loader.get_template('home/product_list.html') return HttpResponse(html_template.render(context, request)) except Exception: html_template = loader.get_template('home/product_list.html') return HttpResponse(html_template.render(context, request)) @login_required(login_url="/login/") def sensor_data_list(request, sensor_name): color_form = handle_color_form(request) profile, created = Profile.objects.get_or_create(user=request.user) background_color = handle_background_color() sensor = get_object_or_404(Sensor, name=sensor_name) sensor_data_list = SensorData.objects.filter(name=sensor).order_by('-timestamp') paginator = Paginator(sensor_data_list, 20) # 每頁顯示 20 筆資料 page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) context = { 'sensor': sensor, 'page_obj': page_obj, 'background_color': background_color, 'color_form' : color_form, 'profile': profile, } return render(request, 'home/sensor_data_list.html', context) @login_required(login_url="/login/") def add_sensor_data(request): sensor_name = request.GET.get('name') data = request.GET.get('data') if not sensor_name: return HttpResponse("Sensor name is required", status=400) if not data: return HttpResponse("Data is required", status=400) try: float_data = round(float(data), 2) except ValueError: return HttpResponse("Data must be a valid number", status=400) try: sensor = Sensor.objects.get(name=sensor_name) SensorData.objects.create(name=sensor, data=float_data) return HttpResponse("Data added successfully", status=200) except Sensor.DoesNotExist: return HttpResponse("Sensor not found", status=404) except Exception as e: return HttpResponse(f"An error occurred: {e}", status=500) @login_required(login_url="/login/") def delete_sensor(request, sensor_name): try: color_form = handle_color_form(request) background_color = handle_background_color() profile, created = Profile.objects.get_or_create(user=request.user) context = { "background_color": background_color, "color_form" : color_form, "profile" : profile } sensor = get_object_or_404(Sensor, name=sensor_name) sensor.delete() html_template = loader.get_template('home/index.html') return HttpResponse(html_template.render(context, request)) except Exception: html_template = loader.get_template('home/index.html') return HttpResponse(html_template.render(context, request)) @login_required(login_url="/login/") def get_sensors(request): sensors = Sensor.objects.all().values('name') return JsonResponse(list(sensors), safe=False) @login_required(login_url="/login/") def get_sensor_data(request, sensor_name): now = datetime.now() today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) data = SensorData.objects.filter(name__name=sensor_name, timestamp__gte=today_start).values('timestamp', 'data') sensor = get_object_or_404(Sensor, name=sensor_name) today = datetime.now().date() start_date = today - timedelta(days=5) historical_data = SensorData.objects.filter(name=sensor, timestamp__date__range=(start_date, today)) max_min_data = [] for single_date in (start_date + timedelta(n) for n in range(6)): daily_data = historical_data.filter(timestamp__date=single_date) if daily_data.exists(): daily_data_numeric = [float(d) for d in daily_data.values_list('data', flat=True)] max_min_data.append({ 'date': single_date.strftime('%Y-%m-%d'), 'max': max(daily_data_numeric), 'min': min(daily_data_numeric), 'count': daily_data.count(), }) else: max_min_data.append({ 'date': single_date.strftime('%Y-%m-%d'), 'max': None, 'min': None, 'count': 0, }) response_data = { 'current_data': list(data), 'historical_data': max_min_data } return JsonResponse(response_data, safe=False) #-----------------API介面-----------------# def get_counts(request): sensor_count = Sensor.objects.count() product_count = Product.objects.count() return JsonResponse({'platform': '生產端平台', 'sensor_count': sensor_count, 'product_count': product_count}) def get_sensor_names(request): sensor_names = list(Sensor.objects.values_list('name', flat=True)) return JsonResponse(sensor_names, safe=False) def get_sensor_data_by_name(request, sensor_name, count): count = int(request.GET.get('count', 10)) # 預設為10筆 sensor_data = SensorData.objects.filter(name__name=sensor_name).order_by('-timestamp')[:count] data = list(sensor_data.values('timestamp', 'data')) return JsonResponse(data, safe=False) def get_product_ids(request): product_ids = list(Product.objects.values_list('id', flat=True)) return JsonResponse(product_ids, safe=False) def get_product_details(request, product_id): try: product = Product.objects.get(id=product_id) product_images = ProductImage.objects.filter(product=product) image_urls = [image.image.url for image in product_images] image_urls.insert(0, product.image.url) product_data = { 'name': product.name, 'email': product.email, 'phone_number': product.phone_number, 'address': product.address, 'product_name': product.product_name, 'price': product.price, 'description': product.description, 'images': image_urls } return JsonResponse(product_data, safe=False) except Product.DoesNotExist: return JsonResponse({'error': 'Product not found'}, status=404) def custom_404(request, exception): return render(request, 'home/page-404.html', {}, status=404) def custom_403(request, exception): return render(request, 'home/page-403.html', {}, status=403) def custom_500(request): return render(request, 'home/page-500.html', {}, status=500)