1.安裝套件
pip install django
pip install paho-mqtt
2.建立mqtttest專案
django-admin startproject mqtttest
3.進入專案
cd mqtttest
4.建立app
python manage.py startapp mqttapp
5.修改mqtttest/mqtttest/settings.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 | """
Django settings for mqtttest project.
Generated by 'django-admin startproject' using Django 4.2.9.
For more information on this file, see
https://docs.djangoproject.com/en/4.2/topics/settings/
For the full list of settings and their values, see
https://docs.djangoproject.com/en/4.2/ref/settings/
"""
from pathlib import Path
# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/4.2/howto/deployment/checklist/
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = 'django-insecure-q2ia=(7x3d5_mxao#-sl)+)crp9j%e*732fg4jhid-f&sln5k('
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True
ALLOWED_HOSTS = []
# Application definition
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'mqttapp',
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
ROOT_URLCONF = 'mqtttest.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [BASE_DIR / 'templates'],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
WSGI_APPLICATION = 'mqtttest.wsgi.application'
# Database
# https://docs.djangoproject.com/en/4.2/ref/settings/#databases
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': BASE_DIR / 'db.sqlite3',
}
}
# Password validation
# https://docs.djangoproject.com/en/4.2/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
},
]
# Internationalization
# https://docs.djangoproject.com/en/4.2/topics/i18n/
LANGUAGE_CODE = 'en-us'
TIME_ZONE = 'UTC'
USE_I18N = True
USE_TZ = True
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/4.2/howto/static-files/
STATIC_URL = 'static/'
# Default primary key field type
# https://docs.djangoproject.com/en/4.2/ref/settings/#default-auto-field
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
MQTT_SERVER = 'broker.MQTTGO.io'
MQTT_PORT = 1883
MQTT_KEEPALIVE = 60
MQTT_USER = ''
MQTT_PASSWORD = ''
|
6.修改mqtttest/mqtttest/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 | """
URL configuration for mqtttest project.
The `urlpatterns` list routes URLs to views. For more information please see:
https://docs.djangoproject.com/en/4.2/topics/http/urls/
Examples:
Function views
1. Add an import: from my_app import views
2. Add a URL to urlpatterns: path('', views.home, name='home')
Class-based views
1. Add an import: from other_app.views import Home
2. Add a URL to urlpatterns: path('', Home.as_view(), name='home')
Including another URLconf
1. Import the include() function: from django.urls import include, path
2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))
"""
from django.contrib import admin
from django.urls import path
from mqttapp.views import home, publish_message, mqtt_messages_json
urlpatterns = [
path('admin/', admin.site.urls),
path('', home),
path('publish/', publish_message, name='publish'),
path('api/messages/', mqtt_messages_json, name='mqtt_messages_json'),
]
|
7.mqtttest/mqttapp/models.py
1
2
3
4
5
6
7
8
9
10
11 | from django.db import models
# Create your models here.
class MQTTMessage(models.Model):
received_at = models.DateTimeField(auto_now_add=True) # 自動紀錄訊息接收時間
topic = models.CharField(max_length=255)
payload = models.TextField()
def __str__(self):
return f"{self.received_at} - {self.topic}"
|
8.mqtttest/mqttapp/admin.py
1
2
3
4
5
6
7
8
9 | from django.contrib import admin
from .models import MQTTMessage
@admin.register(MQTTMessage)
class MQTTMessageAdmin(admin.ModelAdmin):
list_display = ('received_at', 'topic', 'payload') # 可自訂顯示欄位
list_filter = ('topic',)
search_fields = ('topic', 'payload')
ordering = ('-received_at',)
|
9.mqtttest/mqttapp/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 | from django.shortcuts import render
# Create your views here.
import json
from django.http import JsonResponse
from mqttapp.models import MQTTMessage
from django.views.decorators.csrf import csrf_exempt
import paho.mqtt.client as mqtt
def home(request):
return render(request, 'index.html')
@csrf_exempt
def publish_message(request):
if request.method == 'POST':
try:
data = json.loads(request.body)
topic = data.get('topic', 'django/mqtt')
msg = data.get('msg', 'test')
# 即時建立連線並發送
client = mqtt.Client()
client.connect('broker.MQTTGO.io', 1883, 60)
rc, mid = client.publish(topic, msg)
client.disconnect()
return JsonResponse({'code': rc})
except Exception as e:
return JsonResponse({'error': str(e)}, status=400)
return JsonResponse({'error': 'Only POST allowed'}, status=405)
def mqtt_messages_json(request):
messages = MQTTMessage.objects.order_by('-received_at')[:20]
data = [
{
'topic': m.topic,
'payload': m.payload,
'received_at': m.received_at.strftime('%Y-%m-%d %H:%M:%S')
}
for m in messages
]
return JsonResponse(data, safe=False)
|
10.在mqtttest下建立templates資料夾，並新增mqtttest/templates/index.html
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 | <!DOCTYPE html>
<html>
<head>
<title>MQTT 測試</title>
<script>
function loadMessages() {
fetch('/api/messages/')
.then(response => response.json())
.then(data => {
const tbody = document.getElementById('messages');
tbody.innerHTML = '';
data.forEach(item => {
const row = `<tr>
<td>${item.received_at}</td>
<td>${item.topic}</td>
<td>${item.payload}</td>
</tr>`;
tbody.innerHTML += row;
});
});
}
function publishMessage() {
const topic = document.getElementById('topic').value;
const msg = document.getElementById('msg').value;
fetch('/publish/', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ topic, msg })
})
.then(response => response.json())
.then(data => {
alert('已送出 MQTT 訊息！');
})
.catch(error => {
alert('錯誤：' + error);
});
}
setInterval(loadMessages, 3000);
window.onload = loadMessages;
</script>
</head>
<body>
<h1>MQTT 測試</h1>
<h2><a href="https://broker.mqttgo.io/">使用MQTT GO伺服器</a></h2>
<h2>發送 MQTT 訊息</h2>
<label>Topic: <input id="topic" value="django/mqtt"></label><br>
<label>訊息: <input id="msg" value="Hello MQTT!"></label><br>
<button onclick="publishMessage()">送出</button>
<h2>最新接收訊息</h2>
<table border="1">
<thead>
<tr><th>時間</th><th>主題</th><th>訊息</th></tr>
</thead>
<tbody id="messages"></tbody>
</table>
</body>
</html>
|
12.在mqtttest根目錄下建立mqtt_listener.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 | import os
import django
import paho.mqtt.client as mqtt
# 初始化 Django 環境
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mqtttest.settings")
django.setup()
from mqttapp.models import MQTTMessage
from django.conf import settings
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT broker.")
client.subscribe("django/mqtt")
else:
print("Connection failed with code", rc)
def on_message(client, userdata, msg):
payload = msg.payload.decode('utf-8')
print(f"Received message on topic: {msg.topic}, payload: {payload}")
MQTTMessage.objects.create(topic=msg.topic, payload=payload)
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.username_pw_set(settings.MQTT_USER, settings.MQTT_PASSWORD)
client.connect(settings.MQTT_SERVER, settings.MQTT_PORT, settings.MQTT_KEEPALIVE)
client.loop_forever()
|
13.回到命令列，做資料庫同步，並建立超級使用者
python manage.py makemigrations
python manage.py migrate
14.執行網站
python manage.py runserver
15.再打啟另一個命令列
python mqtt_listener.py
16.啟動MQTT GO，連線並訂閱django/mqtt
17.打開瀏覽器進行測試，如最上圖
.png)