pan/HomeOS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: pan:2614ae8880b9edbaa93a6de4cdc08773cdfc1570
Branches Tags
pan:master
...
compare: pan:master
Branches Tags
pan:master

10 Commits
2614ae8880 ... master

Author SHA1 Message Date
walkpan
3ea47d471e feat(规则管理): 添加主题规则自动发现设备功能 
实现设备自动发现规则管理系统，包含以下主要功能：
1. 新增规则管理页面和API，支持创建、编辑和删除主题匹配规则
2. 添加规则匹配引擎，支持+和#通配符匹配设备主题
3. 实现Broker设备注册表，自动发现并管理符合规则的设备
4. 扩展仪表盘显示Broker信息和活跃主题
5. 修改设备卡片和详情页以区分规则发现的设备
6. 添加相关测试用例确保功能稳定性
 2026-03-30 19:28:22 +08:00
walkpan
38766ca792 feat: production build with FastAPI static file serving for SPA 
- Add StaticFiles mount for /assets
- Add catch-all route serving index.html for SPA client routing
- Frontend build: 101KB gzipped total

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 22:05:56 +08:00
walkpan
5f09c6f144 feat: device detail page with controls and logs, broker management page 
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 22:04:49 +08:00
walkpan
4c16e5c0c9 feat: dashboard and devices list pages with cards, toggle, and add device modal 
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 22:02:23 +08:00
walkpan
52fbb1a15a feat: API client, WebSocket composable, and Pinia device store 
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 21:59:50 +08:00
walkpan
f02ff5db0c feat: Vue 3 project scaffolding with Tailwind, Router, Pinia, and Sidebar 
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 21:58:14 +08:00
walkpan
c4d35be54d fix: Windows event loop compatibility, EMQX API response handling, and test fix 
- Add WindowsSelectorEventLoopPolicy for paho-mqtt on Windows
- Handle EMQX /status plain text and /metrics list responses
- Fix test_default_settings to ignore .env file
- Add __main__.py for python -m mqtt_home support

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 21:52:20 +08:00
walkpan
bac099c805 feat: CLI commands for device management, broker monitoring, and serve 
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 21:48:08 +08:00
walkpan
957f489a0d feat: FastAPI app with REST API routes and WebSocket endpoint 
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 21:46:00 +08:00
walkpan
abb170ace6 feat: HA Discovery protocol handler for auto-registering devices 
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
 2026-03-29 21:42:49 +08:00
54 changed files with 6775 additions and 4 deletions
Expand all files Collapse all files

1
.playwright-mcp/console-2026-03-29T15-40-07-523Z.log Normal file
View File

@@ -0,0 +1 @@
[ 71057ms] [ERROR] WebSocket connection to 'ws://localhost:8000/ws/devices' failed: Error in connection establishment: net::ERR_CONNECTION_REFUSED @ http://localhost:8000/assets/devices-D8vTPJoB.js:0

BIN
dashboard-final.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 67 KiB

1367
docs/superpowers/plans/2026-03-29-mqtt-home-frontend.md Normal file
View File

File diff suppressed because it is too large Load Diff

BIN
frontend/device-detail-screenshot.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 42 KiB

12
frontend/index.html Normal file
View File

@@ -0,0 +1,12 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>MQTT Home</title>
</head>
<body class="bg-gray-50 min-h-screen">
<div id="app"></div>
<script type="module" src="/src/main.js"></script>
</body>
</html>

2380
frontend/package-lock.json generated Normal file
View File

File diff suppressed because it is too large Load Diff

24
frontend/package.json Normal file
View File

@@ -0,0 +1,24 @@
{
"name": "mqtt-home-frontend",
"private": true,
"version": "0.1.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "vite build",
"preview": "vite preview"
},
"dependencies": {
"vue": "^3.5.0",
"vue-router": "^4.4.0",
"pinia": "^2.2.0",
"@heroicons/vue": "^2.2.0"
},
"devDependencies": {
"@vitejs/plugin-vue": "^5.1.0",
"vite": "^5.4.0",
"tailwindcss": "^3.4.0",
"postcss": "^8.4.0",
"autoprefixer": "^10.4.0"
}
}

6
frontend/postcss.config.js Normal file
View File

@@ -0,0 +1,6 @@
export default {
plugins: {
tailwindcss: {},
autoprefixer: {},
},
}

12
frontend/src/App.vue Normal file
View File

@@ -0,0 +1,12 @@
<template>
<div class="flex min-h-screen">
<Sidebar />
<main class="flex-1 p-6 overflow-auto">
<router-view />
</main>
</div>
</template>
<script setup>
import Sidebar from './components/Sidebar.vue'
</script>

52
frontend/src/api/index.js Normal file
View File

@@ -0,0 +1,52 @@
const BASE_URL = '/api'
async function request(path, options = {}) {
const url = `${BASE_URL}${path}`
const config = {
headers: {
'Content-Type': 'application/json',
...options.headers,
},
...options,
}
if (config.body && typeof config.body === 'object') {
config.body = JSON.stringify(config.body)
}
const res = await fetch(url, config)
if (!res.ok) {
const error = new Error(`API error: ${res.status}`)
error.status = res.status
try {
error.data = await res.json()
} catch {
error.data = await res.text()
}
throw error
}
if (res.status === 204) return null
return res.json()
}
// Devices
export const getDevices = () => request('/devices')
export const createDevice = (data) => request('/devices', { method: 'POST', body: data })
export const getDevice = (id) => request(`/devices/${id}`)
export const updateDevice = (id, data) => request(`/devices/${id}`, { method: 'PUT', body: data })
export const deleteDevice = (id) => request(`/devices/${id}`, { method: 'DELETE' })
export const sendCommand = (id, payload) =>
request(`/devices/${id}/command`, { method: 'POST', body: { payload } })
export const getDeviceLogs = (id, limit = 20) => request(`/devices/${id}/logs?limit=${limit}`)
// Broker
export const getBrokerStatus = () => request('/broker/status')
export const getBrokerClients = () => request('/broker/clients')
export const getBrokerTopics = () => request('/broker/topics')
// Dashboard
export const getDashboardStats = () => request('/dashboard')
// Rules
export const getRules = () => request('/rules')
export const createRule = (data) => request('/rules', { method: 'POST', body: data })
export const updateRule = (id, data) => request(`/rules/${id}`, { method: 'PUT', body: data })
export const deleteRule = (id) => request(`/rules/${id}`, { method: 'DELETE' })

65
frontend/src/components/AddDeviceModal.vue Normal file
View File

@@ -0,0 +1,65 @@
<template>
<div v-if="show" class="fixed inset-0 bg-black/40 flex items-center justify-center z-50" @click.self="$emit('close')">
<div class="bg-white rounded-xl shadow-xl w-full max-w-md mx-4 p-6">
<h3 class="text-lg font-semibold text-gray-900 mb-4">添加设备</h3>
<form @submit.prevent="submit" class="space-y-4">
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">设备名称</label>
<input v-model="form.name" type="text" required class="w-full border border-gray-300 rounded-lg px-3 py-2 text-sm focus:ring-2 focus:ring-primary-500 focus:border-primary-500" />
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">设备类型</label>
<select v-model="form.type" class="w-full border border-gray-300 rounded-lg px-3 py-2 text-sm focus:ring-2 focus:ring-primary-500 focus:border-primary-500">
<option value="switch">Switch 开关</option>
<option value="light">Light 灯光</option>
<option value="sensor">Sensor 传感器</option>
<option value="binary_sensor">Binary Sensor</option>
<option value="climate">Climate 温控</option>
</select>
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">状态主题 (State Topic)</label>
<input v-model="form.mqtt_topic" type="text" required placeholder="home/living/light" class="w-full border border-gray-300 rounded-lg px-3 py-2 text-sm focus:ring-2 focus:ring-primary-500 focus:border-primary-500" />
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">命令主题 (Command Topic)</label>
<input v-model="form.command_topic" type="text" placeholder="home/living/light/set" class="w-full border border-gray-300 rounded-lg px-3 py-2 text-sm focus:ring-2 focus:ring-primary-500 focus:border-primary-500" />
</div>
<div class="flex justify-end gap-3 pt-2">
<button type="button" @click="$emit('close')" class="px-4 py-2 text-sm font-medium text-gray-700 bg-gray-100 rounded-lg hover:bg-gray-200">取消</button>
<button type="submit" :disabled="submitting" class="px-4 py-2 text-sm font-medium text-white bg-primary-600 rounded-lg hover:bg-primary-700 disabled:opacity-50">添加</button>
</div>
</form>
</div>
</div>
</template>
<script setup>
import { reactive, ref } from 'vue'
import { useDeviceStore } from '../stores/devices'
defineProps({ show: { type: Boolean, default: false } })
defineEmits(['close'])
const store = useDeviceStore()
const submitting = ref(false)
const form = reactive({
name: '',
type: 'switch',
mqtt_topic: '',
command_topic: '',
})
async function submit() {
submitting.value = true
try {
await store.addDevice(form)
form.name = ''
form.mqtt_topic = ''
form.command_topic = ''
} finally {
submitting.value = false
}
}
</script>

78
frontend/src/components/DeviceCard.vue Normal file
View File

@@ -0,0 +1,78 @@
<template>
<div
class="bg-white rounded-xl shadow-sm border border-gray-200 p-4 hover:shadow-md transition-shadow cursor-pointer"
@click="$router.push(`/devices/${device.id}`)"
>
<div class="flex items-start justify-between mb-3">
<div class="flex items-center gap-2">
<span
class="w-2.5 h-2.5 rounded-full"
:class="device.is_online ? 'bg-green-500' : 'bg-gray-300'"
></span>
<span class="text-sm font-medium text-gray-900 truncate max-w-[160px]">{{ device.name }}</span>
</div>
<span class="text-xs px-2 py-0.5 rounded-full bg-gray-100 text-gray-600">{{ device.type }}</span>
</div>
<div class="text-xs text-gray-500 space-y-1">
<div class="flex justify-between">
<span>状态</span>
<span class="font-medium" :class="device.state ? 'text-gray-900' : 'text-gray-400'">
{{ displayState || '未知' }}
</span>
</div>
<div class="flex justify-between">
<span>协议</span>
<span>{{ device.protocol === 'ha_discovery' ? 'HA' : device.protocol === 'topic_rule' ? '规则' : '手动' }}</span>
</div>
</div>
<button
v-if="device.command_topic && canToggle"
class="mt-3 w-full py-1.5 text-sm font-medium rounded-lg transition-colors"
:class="isOn ? 'bg-primary-600 text-white hover:bg-primary-700' : 'bg-gray-100 text-gray-600 hover:bg-gray-200'"
@click.stop="toggle"
>
{{ isOn ? '关闭' : '开启' }}
</button>
</div>
</template>
<script setup>
import { computed } from 'vue'
import { useDeviceStore } from '../stores/devices'
const props = defineProps({
device: { type: Object, required: true },
})
const store = useDeviceStore()
const canToggle = computed(() => ['switch', 'light'].includes(props.device.type))
const isOn = computed(() => {
try {
const s = JSON.parse(props.device.state || '{}')
return s.state === 'ON' || s.state === 'on' || s.state === 'true'
} catch {
return props.device.state === 'ON' || props.device.state === 'on'
}
})
const displayState = computed(() => {
if (!props.device.state) return null
try {
const s = JSON.parse(props.device.state)
return s.state || props.device.state
} catch {
return props.device.state
}
})
async function toggle() {
const newState = isOn.value ? 'off' : 'on'
try {
await store.sendCommand(props.device.id, JSON.stringify({ state: newState }))
} catch {
// silent fail, WebSocket will update state
}
}
</script>

82
frontend/src/components/DeviceControl.vue Normal file
View File

@@ -0,0 +1,82 @@
<template>
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-4">控制</h3>
<div v-if="canToggle" class="mb-4">
<button
@click="toggle"
:disabled="!device.command_topic"
class="px-6 py-2.5 text-sm font-medium rounded-lg transition-colors"
:class="isOn ? 'bg-primary-600 text-white hover:bg-primary-700' : 'bg-gray-100 text-gray-600 hover:bg-gray-200'"
>
{{ isOn ? '关闭' : '开启' }}
</button>
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">发送命令</label>
<div class="flex gap-2">
<input
v-model="commandPayload"
type="text"
placeholder='{"state":"on"}'
:disabled="!device.command_topic"
class="flex-1 border border-gray-300 rounded-lg px-3 py-2 text-sm font-mono focus:ring-2 focus:ring-primary-500 focus:border-primary-500 disabled:bg-gray-50 disabled:text-gray-400"
/>
<button
@click="sendCustom"
:disabled="!device.command_topic || sending"
class="px-4 py-2 text-sm font-medium text-white bg-primary-600 rounded-lg hover:bg-primary-700 disabled:opacity-50"
>
发送
</button>
</div>
<p v-if="!device.command_topic" class="text-xs text-gray-400 mt-1">此设备未配置命令主题</p>
</div>
</div>
</template>
<script setup>
import { computed, ref } from 'vue'
import { useDeviceStore } from '../stores/devices'
const props = defineProps({
device: { type: Object, required: true },
})
const store = useDeviceStore()
const sending = ref(false)
const commandPayload = ref('')
const canToggle = computed(() => ['switch', 'light'].includes(props.device.type))
const isOn = computed(() => {
try {
const s = JSON.parse(props.device.state || '{}')
return s.state === 'ON' || s.state === 'on' || s.state === 'true'
} catch {
return props.device.state === 'ON' || props.device.state === 'on'
}
})
async function toggle() {
const newState = isOn.value ? 'off' : 'on'
sending.value = true
try {
await store.sendCommand(props.device.id, JSON.stringify({ state: newState }))
} finally {
sending.value = false
}
}
async function sendCustom() {
if (!commandPayload.value.trim()) return
sending.value = true
try {
await store.sendCommand(props.device.id, commandPayload.value)
commandPayload.value = ''
} finally {
sending.value = false
}
}
</script>

46
frontend/src/components/DeviceLogList.vue Normal file
View File

@@ -0,0 +1,46 @@
<template>
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-4">消息日志</h3>
<div v-if="logs.length === 0" class="text-gray-400 text-sm">暂无日志</div>
<div v-else class="space-y-1.5 max-h-[300px] overflow-y-auto">
<div
v-for="log in logs"
:key="log.id"
class="flex items-center gap-3 text-sm py-1.5 border-b border-gray-50 last:border-0"
>
<span
class="px-2 py-0.5 rounded text-xs font-medium shrink-0"
:class="log.direction === 'rx' ? 'bg-blue-50 text-blue-700' : 'bg-orange-50 text-orange-700'"
>
{{ log.direction === 'rx' ? 'RX' : 'TX' }}
</span>
<span class="text-gray-700 font-mono text-xs truncate flex-1">{{ log.payload }}</span>
<span class="text-gray-400 text-xs whitespace-nowrap">{{ formatTime(log.timestamp) }}</span>
</div>
</div>
</div>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { getDeviceLogs } from '../api'
const props = defineProps({
deviceId: { type: String, required: true },
})
const logs = ref([])
function formatTime(ts) {
if (!ts) return ''
return new Date(ts).toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit', second: '2-digit' })
}
onMounted(async () => {
try {
logs.value = await getDeviceLogs(props.deviceId)
} catch {
// silent
}
})
</script>

80
frontend/src/components/RuleCard.vue Normal file
View File

@@ -0,0 +1,80 @@
<template>
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4">
<div class="flex items-start justify-between mb-3">
<div class="flex items-center gap-2">
<span
class="w-2.5 h-2.5 rounded-full"
:class="rule.is_enabled ? 'bg-green-500' : 'bg-gray-300'"
></span>
<span class="text-sm font-medium text-gray-900 truncate max-w-[160px]">{{ rule.name }}</span>
</div>
<span class="text-xs px-2 py-0.5 rounded-full bg-gray-100 text-gray-600">{{ rule.device_type }}</span>
</div>
<div class="text-xs text-gray-500 space-y-1 mb-3">
<div class="flex justify-between">
<span>主题模式</span>
<span class="font-mono text-gray-700 truncate max-w-[180px]">{{ rule.topic_pattern }}</span>
</div>
<div v-if="rule.command_template" class="flex justify-between">
<span>命令模板</span>
<span class="font-mono text-gray-700 truncate max-w-[180px]">{{ rule.command_template }}</span>
</div>
<div v-if="rule.state_value_path" class="flex justify-between">
<span>状态路径</span>
<span class="font-mono text-gray-700">{{ rule.state_value_path }}</span>
</div>
</div>
<div class="flex items-center gap-2">
<button
@click="toggleEnabled"
class="flex-1 py-1.5 text-xs font-medium rounded-lg transition-colors"
:class="rule.is_enabled ? 'bg-green-50 text-green-700 hover:bg-green-100' : 'bg-gray-100 text-gray-500 hover:bg-gray-200'"
>
{{ rule.is_enabled ? '已启用' : '已禁用' }}
</button>
<button
@click="$emit('edit', rule)"
class="px-2 py-1.5 text-xs font-medium text-gray-600 bg-gray-50 rounded-lg hover:bg-gray-100"
>
编辑
</button>
<button
@click="handleDelete"
class="px-2 py-1.5 text-xs font-medium text-red-600 bg-red-50 rounded-lg hover:bg-red-100"
>
删除
</button>
</div>
</div>
</template>
<script setup>
import { updateRule, deleteRule } from '../api'
const props = defineProps({
rule: { type: Object, required: true },
})
const emit = defineEmits(['edit', 'deleted'])
async function toggleEnabled() {
try {
await updateRule(props.rule.id, { is_enabled: !props.rule.is_enabled })
props.rule.is_enabled = !props.rule.is_enabled
} catch {
// silent fail
}
}
async function handleDelete() {
if (!confirm(`确定要删除规则「${props.rule.name}」吗？`)) return
try {
await deleteRule(props.rule.id)
emit('deleted', props.rule.id)
} catch {
alert('删除失败')
}
}
</script>

146
frontend/src/components/RuleModal.vue Normal file
View File

@@ -0,0 +1,146 @@
<template>
<div v-if="show" class="fixed inset-0 bg-black/50 flex items-center justify-center z-50" @click.self="$emit('close')">
<div class="bg-white rounded-xl shadow-xl w-full max-w-md mx-4 p-6">
<h3 class="text-lg font-semibold text-gray-900 mb-4">{{ editRule ? '编辑规则' : '添加规则' }}</h3>
<div class="space-y-4">
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">规则名称</label>
<input
v-model="form.name"
type="text"
placeholder="如：客厅开关"
class="w-full px-3 py-2 border border-gray-300 rounded-lg text-sm focus:outline-none focus:ring-2 focus:ring-primary-500"
/>
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">主题模式</label>
<input
v-model="form.topic_pattern"
type="text"
placeholder="如：home/+/state"
class="w-full px-3 py-2 border border-gray-300 rounded-lg text-sm font-mono focus:outline-none focus:ring-2 focus:ring-primary-500"
/>
<p class="text-xs text-gray-400 mt-1">支持 + 匹配单级，# 匹配多级</p>
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">设备类型</label>
<select
v-model="form.device_type"
class="w-full px-3 py-2 border border-gray-300 rounded-lg text-sm focus:outline-none focus:ring-2 focus:ring-primary-500"
>
<option value="switch">开关</option>
<option value="light">灯</option>
<option value="sensor">传感器</option>
<option value="binary_sensor">二进制传感器</option>
<option value="climate">空调/温控</option>
<option value="lock">锁</option>
</select>
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">命令主题模板</label>
<input
v-model="form.command_template"
type="text"
placeholder="如：home/{device_id}/set"
class="w-full px-3 py-2 border border-gray-300 rounded-lg text-sm font-mono focus:outline-none focus:ring-2 focus:ring-primary-500"
/>
<p class="text-xs text-gray-400 mt-1">{device_id} 将替换为主题中匹配的部分</p>
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1">状态值路径</label>
<input
v-model="form.state_value_path"
type="text"
placeholder="如：state（留空则使用整个 payload）"
class="w-full px-3 py-2 border border-gray-300 rounded-lg text-sm font-mono focus:outline-none focus:ring-2 focus:ring-primary-500"
/>
<p class="text-xs text-gray-400 mt-1">JSON 字段路径，留空则使用整个 payload 作为状态</p>
</div>
</div>
<div class="flex justify-end gap-3 mt-6">
<button
@click="$emit('close')"
class="px-4 py-2 text-sm font-medium text-gray-700 bg-gray-100 rounded-lg hover:bg-gray-200"
>
取消
</button>
<button
@click="handleSubmit"
:disabled="!form.name || !form.topic_pattern || submitting"
class="px-4 py-2 text-sm font-medium text-white bg-primary-600 rounded-lg hover:bg-primary-700 disabled:opacity-50"
>
{{ submitting ? '保存中...' : '保存' }}
</button>
</div>
</div>
</div>
</template>
<script setup>
import { ref, watch } from 'vue'
import { createRule, updateRule } from '../api'
const props = defineProps({
show: Boolean,
editRule: Object,
})
const emit = defineEmits(['close', 'saved'])
const form = ref({
name: '',
topic_pattern: '',
device_type: 'switch',
command_template: '',
state_value_path: '',
is_enabled: true,
})
const submitting = ref(false)
watch(() => props.show, (val) => {
if (val && props.editRule) {
form.value = { ...props.editRule }
} else if (val) {
form.value = {
name: '',
topic_pattern: '',
device_type: 'switch',
command_template: '',
state_value_path: '',
is_enabled: true,
}
}
})
async function handleSubmit() {
submitting.value = true
try {
const data = {
name: form.value.name,
topic_pattern: form.value.topic_pattern,
device_type: form.value.device_type,
command_template: form.value.command_template || null,
state_value_path: form.value.state_value_path || null,
is_enabled: form.value.is_enabled,
}
if (props.editRule) {
await updateRule(props.editRule.id, data)
} else {
await createRule(data)
}
emit('saved')
emit('close')
} catch (e) {
alert(e.message || '保存失败')
} finally {
submitting.value = false
}
}
</script>

55
frontend/src/components/Sidebar.vue Normal file
View File

@@ -0,0 +1,55 @@
<template>
<aside class="w-56 bg-white shadow-sm border-r border-gray-200 flex flex-col">
<div class="p-4 border-b border-gray-200">
<h1 class="text-xl font-bold text-primary-700">MQTT Home</h1>
</div>
<nav class="flex-1 p-2 space-y-1">
<router-link
v-for="item in navItems"
:key="item.path"
:to="item.path"
class="flex items-center gap-2 px-3 py-2 rounded-lg text-sm font-medium transition-colors"
:class="isActive(item.path) ? 'bg-primary-50 text-primary-700' : 'text-gray-600 hover:bg-gray-100'"
>
<component :is="item.icon" class="w-5 h-5" />
{{ item.label }}
</router-link>
</nav>
<div class="p-3 border-t border-gray-200">
<div class="flex items-center gap-2 text-xs text-gray-500">
<span class="w-2 h-2 rounded-full" :class="connected ? 'bg-green-500' : 'bg-red-500'"></span>
{{ connected ? 'MQTT 已连接' : 'MQTT 未连接' }}
</div>
</div>
</aside>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { useRoute } from 'vue-router'
import { HomeIcon, CpuChipIcon, Cog6ToothIcon, ServerIcon } from '@heroicons/vue/24/outline'
const route = useRoute()
const connected = ref(false)
const navItems = [
{ path: '/dashboard', label: '仪表盘', icon: HomeIcon },
{ path: '/devices', label: '设备管理', icon: CpuChipIcon },
{ path: '/rules', label: '规则管理', icon: Cog6ToothIcon },
{ path: '/broker', label: 'Broker', icon: ServerIcon },
]
function isActive(path) {
return route.path.startsWith(path) && path !== '/'
}
onMounted(async () => {
try {
const res = await fetch('/health')
const data = await res.json()
connected.value = data.mqtt_connected
} catch {
connected.value = false
}
})
</script>

24
frontend/src/components/StatsCard.vue Normal file
View File

@@ -0,0 +1,24 @@
<template>
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<div class="flex items-center justify-between">
<div>
<p class="text-sm text-gray-500">{{ label }}</p>
<p class="mt-1 text-3xl font-semibold" :class="color">{{ value }}</p>
</div>
<div class="w-12 h-12 rounded-lg flex items-center justify-center" :class="bgColor">
<component :is="icon" class="w-6 h-6" :class="iconColor" />
</div>
</div>
</div>
</template>
<script setup>
defineProps({
label: { type: String, required: true },
value: { type: [Number, String], required: true },
icon: { type: Object, required: true },
color: { type: String, default: 'text-gray-900' },
bgColor: { type: String, default: 'bg-gray-100' },
iconColor: { type: String, default: 'text-gray-600' },
})
</script>

48
frontend/src/composables/useWebSocket.js Normal file
View File

@@ -0,0 +1,48 @@
import { ref, onMounted, onUnmounted } from 'vue'
export function useWebSocket(url, onMessage) {
const connected = ref(false)
let ws = null
let reconnectTimer = null
let retryDelay = 1000
function connect() {
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
const fullUrl = `${protocol}//${window.location.host}${url}`
ws = new WebSocket(fullUrl)
ws.onopen = () => {
connected.value = true
retryDelay = 1000
}
ws.onmessage = (event) => {
try {
const data = JSON.parse(event.data)
onMessage(data)
} catch {
// ignore non-JSON messages
}
}
ws.onclose = () => {
connected.value = false
reconnectTimer = setTimeout(connect, retryDelay)
retryDelay = Math.min(retryDelay * 2, 30000)
}
ws.onerror = () => {
ws.close()
}
}
onMounted(() => connect())
onUnmounted(() => {
clearTimeout(reconnectTimer)
if (ws) ws.close()
})
return { connected }
}

10
frontend/src/main.js Normal file
View File

@@ -0,0 +1,10 @@
import { createApp } from 'vue'
import { createPinia } from 'pinia'
import App from './App.vue'
import router from './router'
import './style.css'
const app = createApp(App)
app.use(createPinia())
app.use(router)
app.mount('#app')

35
frontend/src/router/index.js Normal file
View File

@@ -0,0 +1,35 @@
import { createRouter, createWebHistory } from 'vue-router'
const router = createRouter({
history: createWebHistory(),
routes: [
{ path: '/', redirect: '/dashboard' },
{
path: '/dashboard',
name: 'Dashboard',
component: () => import('../views/DashboardView.vue'),
},
{
path: '/devices',
name: 'Devices',
component: () => import('../views/DevicesView.vue'),
},
{
path: '/devices/:id',
name: 'DeviceDetail',
component: () => import('../views/DeviceDetailView.vue'),
},
{
path: '/rules',
name: 'Rules',
component: () => import('../views/RulesView.vue'),
},
{
path: '/broker',
name: 'Broker',
component: () => import('../views/BrokerView.vue'),
},
],
})
export default router

77
frontend/src/stores/devices.js Normal file
View File

@@ -0,0 +1,77 @@
import { defineStore } from 'pinia'
import { ref } from 'vue'
import { getDevices, createDevice, deleteDevice, sendCommand as apiSendCommand } from '../api'
import { useWebSocket } from '../composables/useWebSocket'
export const useDeviceStore = defineStore('devices', () => {
const devices = ref([])
const loading = ref(false)
const wsConnected = ref(false)
async function fetchDevices() {
loading.value = true
try {
devices.value = await getDevices()
} finally {
loading.value = false
}
}
async function addDevice(data) {
const device = await createDevice(data)
devices.value.unshift(device)
return device
}
async function removeDevice(id) {
await deleteDevice(id)
devices.value = devices.value.filter((d) => d.id !== id)
}
async function sendCommand(id, payload) {
return await apiSendCommand(id, payload)
}
function handleWsMessage(data) {
if (data.type !== 'device_update') return
const idx = devices.value.findIndex((d) => d.id === data.device_id)
if (idx !== -1) {
// Update existing device
const device = devices.value[idx]
devices.value.splice(idx, 1, {
...device,
state: data.state,
is_online: data.is_online,
last_seen: data.last_seen,
})
} else if (data.source === 'broker') {
// New broker device discovered via WebSocket
devices.value.push({
id: data.device_id,
name: data.device_id.replace('broker:', ''),
type: 'switch',
protocol: 'topic_rule',
mqtt_topic: '',
command_topic: null,
state: data.state,
is_online: data.is_online,
last_seen: data.last_seen,
created_at: new Date().toISOString(),
updated_at: new Date().toISOString(),
})
}
}
// Initialize WebSocket
const { connected } = useWebSocket('/ws/devices', handleWsMessage)
return {
devices,
loading,
wsConnected,
fetchDevices,
addDevice,
removeDevice,
sendCommand,
}
})

7
frontend/src/style.css Normal file
View File

@@ -0,0 +1,7 @@
@tailwind base;
@tailwind components;
@tailwind utilities;
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
}

99
frontend/src/views/BrokerView.vue Normal file
View File

@@ -0,0 +1,99 @@
<template>
<div>
<h2 class="text-2xl font-bold text-gray-900 mb-6">Broker 管理</h2>
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5 mb-4">
<h3 class="text-lg font-semibold text-gray-900 mb-3">状态</h3>
<div class="grid grid-cols-2 md:grid-cols-4 gap-4 text-sm">
<div>
<span class="text-gray-500">状态</span>
<p class="font-medium text-green-600">运行中</p>
</div>
<div>
<span class="text-gray-500">连接数</span>
<p class="font-medium">{{ metrics['client.connected'] || 0 }}</p>
</div>
<div>
<span class="text-gray-500">订阅数</span>
<p class="font-medium">{{ metrics['client.subscribe'] || 0 }}</p>
</div>
<div>
<span class="text-gray-500">消息发布</span>
<p class="font-medium">{{ metrics['messages.publish'] || 0 }}</p>
</div>
</div>
<div v-if="statusRaw" class="mt-3 text-xs text-gray-400">{{ statusRaw }}</div>
</div>
<div class="grid grid-cols-1 lg:grid-cols-2 gap-4">
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-3">
已连接客户端 ({{ clients.length }})
</h3>
<div v-if="clients.length === 0" class="text-gray-400 text-sm">暂无客户端</div>
<div v-else class="space-y-2">
<div
v-for="client in clients"
:key="client.clientid"
class="flex items-center justify-between py-1.5 border-b border-gray-50 last:border-0 text-sm"
>
<div class="flex items-center gap-2">
<span
class="w-2 h-2 rounded-full"
:class="client.connected ? 'bg-green-500' : 'bg-gray-300'"
></span>
<span class="font-medium text-gray-900">{{ client.clientid }}</span>
</div>
<span class="text-gray-500 text-xs">{{ client.ip_address }}</span>
</div>
</div>
</div>
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-3">
活跃主题 ({{ topics.length }})
</h3>
<div v-if="topics.length === 0" class="text-gray-400 text-sm">暂无活跃主题</div>
<div v-else class="space-y-1">
<div
v-for="topic in topics"
:key="topic.topic"
class="py-1 border-b border-gray-50 last:border-0 font-mono text-sm text-gray-700"
>
{{ topic.topic }}
</div>
</div>
</div>
</div>
</div>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { getBrokerStatus, getBrokerClients, getBrokerTopics } from '../api'
const metrics = ref({})
const statusRaw = ref('')
const clients = ref([])
const topics = ref([])
onMounted(async () => {
try {
const status = await getBrokerStatus()
metrics.value = status.metrics || {}
statusRaw.value = status.status?.raw || ''
} catch {
// silent
}
try {
clients.value = await getBrokerClients()
} catch {
// silent
}
try {
topics.value = await getBrokerTopics()
} catch {
// silent
}
})
</script>

189
frontend/src/views/DashboardView.vue Normal file
View File

@@ -0,0 +1,189 @@
<template>
<div>
<h2 class="text-2xl font-bold text-gray-900 mb-6">仪表盘</h2>
<!-- Stats Cards -->
<div class="grid grid-cols-2 md:grid-cols-4 gap-4 mb-6">
<StatsCard
label="MQTT 连接"
:value="stats.mqtt_connected ? '已连接' : '未连接'"
:icon="SignalIcon"
:color="stats.mqtt_connected ? 'text-green-600' : 'text-red-600'"
:bg-color="stats.mqtt_connected ? 'bg-green-50' : 'bg-red-50'"
:icon-color="stats.mqtt_connected ? 'text-green-500' : 'text-red-500'"
/>
<StatsCard
label="设备总数"
:value="stats.total_devices"
:icon="CpuChipIcon"
color="text-gray-900"
bg-color="bg-gray-100"
icon-color="text-gray-600"
/>
<StatsCard
label="在线设备"
:value="stats.online_devices"
:icon="WifiIcon"
color="text-green-600"
bg-color="bg-green-50"
icon-color="text-green-500"
/>
<StatsCard
label="离线设备"
:value="stats.offline_devices"
:icon="SignalSlashIcon"
color="text-red-600"
bg-color="bg-red-50"
icon-color="text-red-500"
/>
</div>
<!-- Broker Info + Topics -->
<div class="grid grid-cols-1 lg:grid-cols-2 gap-4 mb-6">
<!-- Broker Info -->
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-4">Broker 信息</h3>
<div v-if="brokerLoading" class="text-gray-400 text-sm">加载中...</div>
<div v-else-if="brokerError" class="text-red-400 text-sm">{{ brokerError }}</div>
<div v-else class="space-y-2 text-sm">
<div class="flex justify-between">
<span class="text-gray-500">状态</span>
<span class="text-green-600">
{{ brokerStatusText }}
</span>
</div>
<div v-if="brokerNode" class="flex justify-between">
<span class="text-gray-500">节点</span>
<span class="text-gray-700 font-mono text-xs truncate max-w-[200px]">{{ brokerNode }}</span>
</div>
<div class="flex justify-between">
<span class="text-gray-500">已连接客户端</span>
<span class="text-gray-700">{{ brokerConnections }}</span>
</div>
<div class="flex justify-between">
<span class="text-gray-500">消息发布数</span>
<span class="text-gray-700">{{ brokerPublish }}</span>
</div>
<div class="flex justify-between">
<span class="text-gray-500">Broker 设备</span>
<span class="text-gray-700">{{ stats.broker_device_count || 0 }}</span>
</div>
</div>
</div>
<!-- Active Topics -->
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-4">活跃主题</h3>
<div v-if="stats.broker_topics && stats.broker_topics.length === 0" class="text-gray-400 text-sm">
暂无活跃主题
</div>
<div v-else class="space-y-1.5">
<div
v-for="topic in stats.broker_topics"
:key="topic"
class="flex items-center gap-2 text-sm py-1 px-2 rounded bg-gray-50"
>
<span class="text-primary-600 font-mono text-xs">#</span>
<span class="font-mono text-gray-700 truncate">{{ topic }}</span>
</div>
</div>
</div>
</div>
<!-- Recent Activity -->
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-4">最近活动</h3>
<div v-if="stats.recent_logs && stats.recent_logs.length === 0" class="text-gray-400 text-sm">
暂无活动
</div>
<div v-else class="space-y-2">
<div
v-for="log in stats.recent_logs"
:key="log.id"
class="flex items-center gap-3 text-sm py-1.5 border-b border-gray-100 last:border-0"
>
<span
class="px-2 py-0.5 rounded text-xs font-medium"
:class="log.direction === 'rx' ? 'bg-blue-50 text-blue-700' : 'bg-orange-50 text-orange-700'"
>
{{ log.direction === 'rx' ? '接收' : '发送' }}
</span>
<span class="text-gray-500 font-mono text-xs truncate max-w-[200px]">{{ log.topic }}</span>
<span class="text-gray-700 truncate flex-1">{{ log.payload }}</span>
<span class="text-gray-400 text-xs whitespace-nowrap">{{ formatTime(log.timestamp) }}</span>
</div>
</div>
</div>
</div>
</template>
<script setup>
import { ref, computed, onMounted } from 'vue'
import { getDashboardStats, getBrokerStatus } from '../api'
import StatsCard from '../components/StatsCard.vue'
import { CpuChipIcon, WifiIcon, SignalSlashIcon, SignalIcon } from '@heroicons/vue/24/outline'
const stats = ref({
total_devices: 0,
online_devices: 0,
offline_devices: 0,
recent_logs: [],
broker_topics: [],
broker_device_count: 0,
broker_online_count: 0,
mqtt_connected: false,
})
const broker = ref(null)
const brokerLoading = ref(false)
const brokerError = ref('')
// Extract display values from real EMQX response format
const brokerStatusText = computed(() => {
if (!broker.value) return '未知'
const raw = broker.value.status?.raw || ''
if (raw.includes('is running') || raw.includes('started')) return '运行中'
return raw || '未知'
})
const brokerNode = computed(() => {
if (!broker.value?.metrics) return ''
return broker.value.metrics.node || ''
})
const brokerConnections = computed(() => {
if (!broker.value?.metrics) return 0
return broker.value.metrics['client.connected'] || 0
})
const brokerPublish = computed(() => {
if (!broker.value?.metrics) return 0
return broker.value.metrics['messages.publish'] || 0
})
function formatTime(ts) {
if (!ts) return ''
const d = new Date(ts)
return d.toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit', second: '2-digit' })
}
onMounted(async () => {
// Fetch dashboard stats
try {
stats.value = await getDashboardStats()
} catch {
// silently fail
}
// Fetch broker status
brokerLoading.value = true
try {
broker.value = await getBrokerStatus()
} catch (e) {
brokerError.value = '无法连接到 Broker'
} finally {
brokerLoading.value = false
}
})
</script>

81
frontend/src/views/DeviceDetailView.vue Normal file
View File

@@ -0,0 +1,81 @@
<template>
<div>
<div v-if="!device" class="text-gray-400">加载中...</div>
<template v-else>
<div class="flex items-center gap-2 mb-6">
<router-link to="/devices" class="text-primary-600 hover:text-primary-700 text-sm">&larr; 返回设备列表</router-link>
</div>
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5 mb-4">
<div class="flex items-start justify-between">
<div>
<div class="flex items-center gap-2 mb-1">
<h2 class="text-xl font-bold text-gray-900">{{ device.name }}</h2>
<span
class="w-2.5 h-2.5 rounded-full"
:class="device.is_online ? 'bg-green-500' : 'bg-gray-300'"
></span>
</div>
<div class="text-sm text-gray-500 space-x-3">
<span>{{ device.type }}</span>
<span>{{ device.protocol === 'ha_discovery' ? 'HA Discovery' : device.protocol === 'topic_rule' ? '主题规则' : '自定义' }}</span>
<span>ID: {{ device.id.slice(0, 8) }}</span>
</div>
</div>
<button
v-if="device.protocol !== 'topic_rule'"
@click="handleDelete"
class="px-3 py-1.5 text-sm font-medium text-red-600 bg-red-50 rounded-lg hover:bg-red-100"
>
删除设备
</button>
</div>
<div class="mt-4 grid grid-cols-2 gap-x-8 gap-y-2 text-sm">
<div><span class="text-gray-500">状态主题：</span><span class="font-mono text-gray-700">{{ device.mqtt_topic }}</span></div>
<div><span class="text-gray-500">命令主题：</span><span class="font-mono text-gray-700">{{ device.command_topic || '无' }}</span></div>
<div><span class="text-gray-500">当前状态：</span><span class="font-mono text-gray-700">{{ device.state || '无' }}</span></div>
<div><span class="text-gray-500">最后活跃：</span><span class="text-gray-700">{{ device.last_seen || '从未' }}</span></div>
</div>
</div>
<div class="grid grid-cols-1 lg:grid-cols-2 gap-4">
<DeviceControl :device="device" />
<template v-if="device.protocol === 'topic_rule'">
<div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
<h3 class="text-lg font-semibold text-gray-900 mb-2">设备信息</h3>
<p class="text-sm text-gray-500">
此设备通过主题规则自动发现，由规则管理系统管理。
</p>
</div>
</template>
<DeviceLogList v-else :device-id="device.id" />
</div>
</template>
</div>
</template>
<script setup>
import { ref, onMounted, computed } from 'vue'
import { useRoute, useRouter } from 'vue-router'
import { useDeviceStore } from '../stores/devices'
import DeviceControl from '../components/DeviceControl.vue'
import DeviceLogList from '../components/DeviceLogList.vue'
const route = useRoute()
const router = useRouter()
const store = useDeviceStore()
const device = computed(() => store.devices.find((d) => d.id === route.params.id))
onMounted(() => {
if (!device.value) store.fetchDevices()
})
async function handleDelete() {
if (!confirm('确定要删除此设备吗？')) return
await store.removeDevice(route.params.id)
router.push('/devices')
}
</script>

38
frontend/src/views/DevicesView.vue Normal file
View File

@@ -0,0 +1,38 @@
<template>
<div>
<div class="flex items-center justify-between mb-6">
<h2 class="text-2xl font-bold text-gray-900">设备管理</h2>
<button
@click="showAdd = true"
class="px-4 py-2 text-sm font-medium text-white bg-primary-600 rounded-lg hover:bg-primary-700"
>
+ 添加设备
</button>
</div>
<div v-if="store.loading" class="text-gray-400">加载中...</div>
<div v-else-if="store.devices.length === 0" class="text-center py-16 text-gray-400">
<p>暂无设备</p>
<p class="text-sm mt-1">配置主题匹配规则自动发现设备，或通过 HA Discovery 自动发现</p>
</div>
<div v-else class="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 xl:grid-cols-4 gap-4">
<DeviceCard v-for="device in store.devices" :key="device.id" :device="device" />
</div>
<AddDeviceModal :show="showAdd" @close="showAdd = false" />
</div>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { useDeviceStore } from '../stores/devices'
import DeviceCard from '../components/DeviceCard.vue'
import AddDeviceModal from '../components/AddDeviceModal.vue'
const store = useDeviceStore()
const showAdd = ref(false)
onMounted(() => store.fetchDevices())
</script>

69
frontend/src/views/RulesView.vue Normal file
View File

@@ -0,0 +1,69 @@
<template>
<div>
<div class="flex items-center justify-between mb-6">
<h2 class="text-2xl font-bold text-gray-900">规则管理</h2>
<button
@click="showAdd = true"
class="px-4 py-2 text-sm font-medium text-white bg-primary-600 rounded-lg hover:bg-primary-700"
>
+ 添加规则
</button>
</div>
<div v-if="loading" class="text-gray-400">加载中...</div>
<div v-else-if="rules.length === 0" class="text-center py-16 text-gray-400">
<p>暂无规则</p>
<p class="text-sm mt-1">添加主题匹配规则，系统将自动发现 MQTT 设备</p>
</div>
<div v-else class="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 xl:grid-cols-4 gap-4">
<RuleCard
v-for="rule in rules"
:key="rule.id"
:rule="rule"
@edit="editRule"
@deleted="handleDeleted"
/>
</div>
<RuleModal :show="showAdd" :edit-rule="editRuleData" @close="closeModal" @saved="fetchRules" />
</div>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { getRules } from '../api'
import RuleCard from '../components/RuleCard.vue'
import RuleModal from '../components/RuleModal.vue'
const rules = ref([])
const loading = ref(false)
const showAdd = ref(false)
const editRuleData = ref(null)
async function fetchRules() {
loading.value = true
try {
rules.value = await getRules()
} finally {
loading.value = false
}
}
function editRule(rule) {
editRuleData.value = rule
showAdd.value = true
}
function closeModal() {
showAdd.value = false
editRuleData.value = null
}
function handleDeleted(ruleId) {
rules.value = rules.value.filter((r) => r.id !== ruleId)
}
onMounted(() => fetchRules())
</script>

18
frontend/tailwind.config.js Normal file
View File

@@ -0,0 +1,18 @@
/** @type {import('tailwindcss').Config} */
export default {
content: ['./index.html', './src/**/*.{vue,js,ts}'],
theme: {
extend: {
colors: {
primary: {
50: '#eff6ff',
100: '#dbeafe',
500: '#3b82f6',
600: '#2563eb',
700: '#1d4ed8',
},
},
},
},
plugins: [],
}

20
frontend/vite.config.js Normal file
View File

@@ -0,0 +1,20 @@
import { defineConfig } from 'vite'
import vue from '@vitejs/plugin-vue'
export default defineConfig({
plugins: [vue()],
server: {
port: 3000,
proxy: {
'/api': 'http://localhost:8000',
'/ws': {
target: 'http://localhost:8000',
ws: true,
},
'/health': 'http://localhost:8000',
},
},
build: {
outDir: 'dist',
},
})

3
src/mqtt_home/__main__.py Normal file
View File

@@ -0,0 +1,3 @@
from mqtt_home.cli import cli
cli()

11
src/mqtt_home/api/__init__.py Normal file
View File

@@ -0,0 +1,11 @@
from fastapi import APIRouter
from mqtt_home.api.devices import router as devices_router
from mqtt_home.api.broker import router as broker_router
from mqtt_home.api.dashboard import router as dashboard_router
from mqtt_home.api.rules import router as rules_router
api_router = APIRouter(prefix="/api")
api_router.include_router(devices_router, prefix="/devices", tags=["devices"])
api_router.include_router(broker_router, prefix="/broker", tags=["broker"])
api_router.include_router(dashboard_router, prefix="/dashboard", tags=["dashboard"])
api_router.include_router(rules_router, prefix="/rules", tags=["rules"])

43
src/mqtt_home/api/broker.py Normal file
View File

@@ -0,0 +1,43 @@
from fastapi import APIRouter, HTTPException
from mqtt_home.schemas import BrokerClient, BrokerTopic
router = APIRouter()
@router.get("/status")
async def broker_status():
from mqtt_home.main import app
emqx = getattr(app.state, "emqx_client", None)
if not emqx:
raise HTTPException(status_code=503, detail="EMQX API client not configured")
try:
status = await emqx.get_broker_status()
metrics = await emqx.get_metrics()
return {"status": status, "metrics": metrics}
except Exception as e:
raise HTTPException(status_code=502, detail=f"EMQX API error: {e}")
@router.get("/clients", response_model=list[BrokerClient])
async def broker_clients(limit: int = 100):
from mqtt_home.main import app
emqx = getattr(app.state, "emqx_client", None)
if not emqx:
raise HTTPException(status_code=503, detail="EMQX API client not configured")
try:
return await emqx.get_clients(limit)
except Exception as e:
raise HTTPException(status_code=502, detail=f"EMQX API error: {e}")
@router.get("/topics", response_model=list[BrokerTopic])
async def broker_topics(limit: int = 100):
from mqtt_home.main import app
emqx = getattr(app.state, "emqx_client", None)
if not emqx:
raise HTTPException(status_code=503, detail="EMQX API client not configured")
try:
return await emqx.get_topics(limit)
except Exception as e:
raise HTTPException(status_code=502, detail=f"EMQX API error: {e}")

45
src/mqtt_home/api/dashboard.py Normal file
View File

@@ -0,0 +1,45 @@
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from mqtt_home.database import get_db
from mqtt_home.device_registry import get_dashboard_stats
from mqtt_home.schemas import DashboardStats
from mqtt_home.emqx_api import EmqxApiClient
router = APIRouter()
@router.get("")
async def dashboard(db: AsyncSession = Depends(get_db)):
from mqtt_home.main import app
stats = await get_dashboard_stats(db)
# Add broker device counts
registry = getattr(app.state, "broker_registry", None)
broker_devices = registry.get_all() if registry else []
broker_online = sum(1 for d in broker_devices if d.is_online)
stats["total_devices"] += len(broker_devices)
stats["online_devices"] += broker_online
stats["offline_devices"] += len(broker_devices) - broker_online
# Add broker topics from EMQX API
emqx = getattr(app.state, "emqx_client", None)
broker_topics = []
if emqx:
try:
topics = await emqx.get_topics()
broker_topics = [t.get("topic", "") for t in topics]
except Exception:
pass
stats["broker_topics"] = broker_topics
stats["broker_device_count"] = len(broker_devices)
stats["broker_online_count"] = broker_online
# Add MQTT connection status
mqtt = getattr(app.state, "mqtt_client", None)
stats["mqtt_connected"] = mqtt.is_connected if mqtt else False
return stats

122
src/mqtt_home/api/devices.py Normal file
View File

@@ -0,0 +1,122 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from mqtt_home.database import get_db
from mqtt_home.device_registry import (
list_devices, get_device, create_device, update_device,
delete_device, send_command, get_device_logs,
)
from mqtt_home.models import DeviceLog
from mqtt_home.schemas import (
DeviceCreate, DeviceUpdate, DeviceCommand,
DeviceResponse, DeviceLogResponse,
)
router = APIRouter()
@router.get("", response_model=list[DeviceResponse])
async def get_devices(db: AsyncSession = Depends(get_db)):
db_devices = await list_devices(db)
result = [DeviceResponse.model_validate(d) for d in db_devices]
# Merge broker devices
from mqtt_home.main import app
registry = getattr(app.state, "broker_registry", None)
if registry:
for bd in registry.get_all():
result.append(DeviceResponse(
id=bd.id,
name=bd.name,
type=bd.type,
protocol=bd.protocol,
mqtt_topic=bd.mqtt_topic,
command_topic=bd.command_topic,
state=bd.state,
is_online=bd.is_online,
last_seen=bd.last_seen,
created_at=bd.created_at,
updated_at=bd.updated_at,
))
return result
@router.post("", response_model=DeviceResponse, status_code=201)
async def add_device(data: DeviceCreate, db: AsyncSession = Depends(get_db)):
return await create_device(db, data)
@router.get("/{device_id}", response_model=DeviceResponse)
async def get_device_detail(device_id: str, db: AsyncSession = Depends(get_db)):
device = await get_device(db, device_id)
if not device:
raise HTTPException(status_code=404, detail="Device not found")
return device
@router.put("/{device_id}", response_model=DeviceResponse)
async def patch_device(device_id: str, data: DeviceUpdate, db: AsyncSession = Depends(get_db)):
device = await update_device(db, device_id, data)
if not device:
raise HTTPException(status_code=404, detail="Device not found")
return device
@router.delete("/{device_id}", status_code=204)
async def remove_device(device_id: str, db: AsyncSession = Depends(get_db)):
if not await delete_device(db, device_id):
raise HTTPException(status_code=404, detail="Device not found")
@router.post("/{device_id}/command", response_model=DeviceLogResponse)
async def command_device(
device_id: str, data: DeviceCommand,
db: AsyncSession = Depends(get_db),
):
from mqtt_home.main import app
mqtt_client = getattr(app.state, "mqtt_client", None)
if not mqtt_client or not mqtt_client.is_connected:
raise HTTPException(status_code=503, detail="MQTT not connected")
# Handle broker devices
if device_id.startswith("broker:"):
registry = getattr(app.state, "broker_registry", None)
if not registry:
raise HTTPException(status_code=404, detail="Device not found")
bd = registry.get(device_id)
if not bd:
raise HTTPException(status_code=404, detail="Device not found")
if not bd.command_topic:
raise HTTPException(status_code=400, detail="Device has no command_topic configured")
await mqtt_client.publish(bd.command_topic, data.payload)
# Create a log entry in DB for broker device commands
log = DeviceLog(
device_id=device_id,
direction="tx",
topic=bd.command_topic,
payload=data.payload,
)
db.add(log)
await db.commit()
await db.refresh(log)
return log
# Original DB device handling
try:
log = await send_command(
db, device_id, data.payload,
publish_fn=mqtt_client.publish,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
if not log:
raise HTTPException(status_code=404, detail="Device not found")
return log
@router.get("/{device_id}/logs", response_model=list[DeviceLogResponse])
async def read_device_logs(device_id: str, limit: int = 20, db: AsyncSession = Depends(get_db)):
return await get_device_logs(db, device_id, limit)

65
src/mqtt_home/api/rules.py Normal file
View File

@@ -0,0 +1,65 @@
import logging
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from mqtt_home.database import get_db
from mqtt_home.rule_registry import list_rules, get_rule, create_rule, update_rule, delete_rule
from mqtt_home.schemas import RuleCreate, RuleUpdate, RuleResponse
logger = logging.getLogger(__name__)
router = APIRouter()
async def _refresh_rules_cache():
"""Refresh the cached topic rules in app state."""
from mqtt_home.main import app
from mqtt_home.database import get_session_factory
session_factory = get_session_factory()
async with session_factory() as db:
app.state.topic_rules = [r for r in await list_rules(db) if r.is_enabled]
@router.get("", response_model=list[RuleResponse])
async def get_rules(db: AsyncSession = Depends(get_db)):
return await list_rules(db)
@router.post("", response_model=RuleResponse, status_code=201)
async def add_rule(data: RuleCreate, db: AsyncSession = Depends(get_db)):
rule = await create_rule(db, data)
await _refresh_rules_cache()
return rule
@router.get("/{rule_id}", response_model=RuleResponse)
async def get_rule_detail(rule_id: int, db: AsyncSession = Depends(get_db)):
rule = await get_rule(db, rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Rule not found")
return rule
@router.put("/{rule_id}", response_model=RuleResponse)
async def patch_rule(rule_id: int, data: RuleUpdate, db: AsyncSession = Depends(get_db)):
rule = await update_rule(db, rule_id, data)
if not rule:
raise HTTPException(status_code=404, detail="Rule not found")
await _refresh_rules_cache()
return rule
@router.delete("/{rule_id}", status_code=204)
async def remove_rule(rule_id: int, db: AsyncSession = Depends(get_db)):
if not await delete_rule(db, rule_id):
raise HTTPException(status_code=404, detail="Rule not found")
# Clean up broker devices associated with this rule
from mqtt_home.main import app
registry = getattr(app.state, "broker_registry", None)
if registry:
removed = registry.remove_by_rule(rule_id)
if removed:
logger.info("Removed %d broker devices for rule %d", len(removed), rule_id)
await _refresh_rules_cache()

107
src/mqtt_home/broker_devices.py Normal file
View File

@@ -0,0 +1,107 @@
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional
from mqtt_home.topic_matcher import match_topic, extract_device_id, build_command_topic, extract_state_value
logger = logging.getLogger(__name__)
@dataclass
class BrokerDevice:
"""In-memory broker device, not persisted to DB"""
id: str # "broker:{device_id}" e.g. "broker:fire"
name: str # Human-readable name, e.g. "fire"
type: str # Device type from rule, e.g. "switch"
protocol: str = "topic_rule"
mqtt_topic: str = "" # The matched topic, e.g. "home/fire"
command_topic: Optional[str] = None # Built from template, e.g. "home/fire/set"
state: Optional[str] = None # Latest payload (or extracted value)
is_online: bool = False
last_seen: Optional[datetime] = None
rule_id: int = 0
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
class BrokerDeviceRegistry:
"""In-memory registry for devices discovered via topic format rules"""
def __init__(self):
self._devices: dict[str, BrokerDevice] = {}
def update_or_create(self, topic: str, payload: str, rule) -> Optional[BrokerDevice]:
"""Try to match topic against rule pattern, create/update device if matched.
Args:
topic: The MQTT topic that received a message
payload: The message payload string
rule: TopicFormatRule instance (SQLAlchemy model or dict-like)
Returns:
Updated or created BrokerDevice, or None if topic doesn't match rule
"""
# 1. Check if topic matches the rule's pattern
pattern = rule.topic_pattern
if not match_topic(topic, pattern):
return None
# 2. Extract device_id from topic
device_id_raw = extract_device_id(topic, pattern)
if not device_id_raw:
return None
device_id = f"broker:{device_id_raw}"
# 3. Extract state value from payload
state = extract_state_value(payload, rule.state_value_path)
# 4. Build command topic from template
command_topic = build_command_topic(rule.command_template, device_id_raw)
now = datetime.now(timezone.utc)
# 5. Update existing or create new
if device_id in self._devices:
device = self._devices[device_id]
device.state = state
device.is_online = True
device.last_seen = now
device.updated_at = now
# Update command_topic in case rule changed
if command_topic:
device.command_topic = command_topic
else:
device = BrokerDevice(
id=device_id,
name=device_id_raw,
type=rule.device_type,
mqtt_topic=topic,
command_topic=command_topic,
state=state,
is_online=True,
last_seen=now,
rule_id=rule.id,
)
self._devices[device_id] = device
logger.info("Broker device discovered: %s (topic=%s, rule=%d)", device_id, topic, rule.id)
return device
def get_all(self) -> list[BrokerDevice]:
return list(self._devices.values())
def get(self, device_id: str) -> Optional[BrokerDevice]:
return self._devices.get(device_id)
def remove_by_rule(self, rule_id: int) -> list[str]:
"""Remove all devices associated with a rule. Returns list of removed device IDs."""
to_remove = [did for did, d in self._devices.items() if d.rule_id == rule_id]
for did in to_remove:
del self._devices[did]
return to_remove
def clear(self):
self._devices.clear()

253
src/mqtt_home/cli.py Normal file
View File

@@ -0,0 +1,253 @@
import asyncio
import sys
import click
# Windows compatibility: use SelectorEventLoop for paho-mqtt
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
from mqtt_home.config import get_settings
from mqtt_home.database import init_db, get_session_factory
from mqtt_home.emqx_api import EmqxApiClient
from mqtt_home.device_registry import (
list_devices, get_device, create_device, delete_device,
send_command, get_device_logs,
)
from mqtt_home.schemas import DeviceCreate
from mqtt_home.mqtt_client import MqttClient
def run_async(coro):
asyncio.run(coro)
@click.group()
def cli():
"""MQTT 智能家居管理工具"""
pass
@cli.group()
def device():
"""设备管理"""
pass
@device.command("list")
def device_list():
"""列出所有设备"""
async def _run():
await init_db()
factory = get_session_factory()
async with factory() as db:
devices = await list_devices(db)
if not devices:
click.echo("暂无设备")
return
for d in devices:
status = "ON " if d.is_online else "OFF"
state = d.state or "-"
click.echo(f"[{status}] [{d.id[:8]}] {d.name} ({d.type}) state={state}")
run_async(_run())
@device.command("add")
@click.option("--name", required=True, help="设备名称")
@click.option("--type", "device_type", default="switch", help="设备类型")
@click.option("--state-topic", required=True, help="状态主题")
@click.option("--command-topic", default=None, help="命令主题")
def device_add(name, device_type, state_topic, command_topic):
"""手动添加设备"""
async def _run():
await init_db()
factory = get_session_factory()
async with factory() as db:
d = await create_device(db, DeviceCreate(
name=name,
type=device_type,
mqtt_topic=state_topic,
command_topic=command_topic,
))
click.echo(f"设备已创建: {d.id} - {d.name}")
run_async(_run())
@device.command("info")
@click.argument("device_id")
def device_info(device_id):
"""查看设备详情"""
async def _run():
await init_db()
factory = get_session_factory()
async with factory() as db:
d = await get_device(db, device_id)
if not d:
click.echo(f"设备不存在: {device_id}", err=True)
return
click.echo(f"ID: {d.id}")
click.echo(f"名称: {d.name}")
click.echo(f"类型: {d.type}")
click.echo(f"协议: {d.protocol}")
click.echo(f"状态主题: {d.mqtt_topic}")
click.echo(f"命令主题: {d.command_topic or '无'}")
click.echo(f"当前状态: {d.state or '无'}")
click.echo(f"在线: {'是' if d.is_online else '否'}")
click.echo(f"最后活跃: {d.last_seen or '从未'}")
run_async(_run())
@device.command("remove")
@click.argument("device_id")
def device_remove(device_id):
"""删除设备"""
async def _run():
await init_db()
factory = get_session_factory()
async with factory() as db:
if await delete_device(db, device_id):
click.echo(f"设备已删除: {device_id}")
else:
click.echo(f"设备不存在: {device_id}", err=True)
run_async(_run())
@device.command("command")
@click.argument("device_id")
@click.option("--payload", required=True, help="命令内容（JSON）")
def device_command(device_id, payload):
"""向设备发送命令"""
async def _run():
settings = get_settings()
await init_db()
factory = get_session_factory()
import aiomqtt
async with aiomqtt.Client(
hostname=settings.mqtt_host,
port=settings.mqtt_port,
username=settings.mqtt_username or None,
password=settings.mqtt_password or None,
) as client:
async def publish_fn(topic, p):
await client.publish(topic, p, qos=1)
async with factory() as db:
try:
log = await send_command(db, device_id, payload, publish_fn=publish_fn)
if log:
click.echo(f"命令已发送到 {log.topic}: {log.payload}")
else:
click.echo(f"设备不存在: {device_id}", err=True)
except ValueError as e:
click.echo(str(e), err=True)
run_async(_run())
@device.command("logs")
@click.argument("device_id")
@click.option("--limit", default=20, help="日志条数")
def device_logs(device_id, limit):
"""查看设备消息日志"""
async def _run():
await init_db()
factory = get_session_factory()
async with factory() as db:
logs = await get_device_logs(db, device_id, limit)
if not logs:
click.echo("暂无日志")
return
for log in logs:
direction = "RX" if log.direction == "rx" else "TX"
click.echo(f"[{log.timestamp}] {direction} {log.topic} | {log.payload}")
run_async(_run())
@cli.group()
def broker():
"""Broker 管理"""
pass
@broker.command("status")
def broker_status():
"""查看 Broker 状态"""
async def _run():
settings = get_settings()
emqx = EmqxApiClient(settings)
try:
status = await emqx.get_broker_status()
click.echo(f"版本: {status.get('version', 'unknown')}")
click.echo(f"运行时间: {status.get('uptime', 0)}s")
metrics = await emqx.get_metrics()
click.echo(f"连接数: {metrics.get('connections.count', 0)}")
click.echo(f"订阅数: {metrics.get('subscriptions.count', 0)}")
click.echo(f"主题数: {metrics.get('topics.count', 0)}")
except Exception as e:
click.echo(f"连接 EMQX 失败: {e}", err=True)
finally:
await emqx.close()
run_async(_run())
@broker.command("clients")
def broker_clients():
"""列出已连接客户端"""
async def _run():
settings = get_settings()
emqx = EmqxApiClient(settings)
try:
clients = await emqx.get_clients()
if not clients:
click.echo("暂无客户端连接")
return
for c in clients:
status = "在线" if c.get("connected") else "离线"
click.echo(f"[{c.get('clientid')}] {status} ip={c.get('ip_address')}")
except Exception as e:
click.echo(f"连接 EMQX 失败: {e}", err=True)
finally:
await emqx.close()
run_async(_run())
@broker.command("topics")
def broker_topics():
"""列出活跃主题"""
async def _run():
settings = get_settings()
emqx = EmqxApiClient(settings)
try:
topics = await emqx.get_topics()
if not topics:
click.echo("暂无活跃主题")
return
for t in topics:
click.echo(t.get("topic", ""))
except Exception as e:
click.echo(f"连接 EMQX 失败: {e}", err=True)
finally:
await emqx.close()
run_async(_run())
@cli.command("serve")
def serve():
"""启动 Web 服务和 MQTT 客户端"""
import uvicorn
from mqtt_home.config import get_settings
settings = get_settings()
uvicorn.run(
"mqtt_home.main:app",
host=settings.web_host,
port=settings.web_port,
reload=False,
)

84
src/mqtt_home/discovery.py Normal file
View File

@@ -0,0 +1,84 @@
import json
import logging
from sqlalchemy.ext.asyncio import AsyncSession
from mqtt_home.models import Device
from mqtt_home.mqtt_client import MqttClient
from mqtt_home.device_registry import create_device, handle_state_update
logger = logging.getLogger(__name__)
DISCOVERY_TOPIC_PREFIX = "homeassistant/"
def parse_discovery_topic(topic: str) -> dict[str, str] | None:
"""Parse HA Discovery topic, returns component and node_id"""
parts = topic.split("/")
if len(parts) < 4 or parts[0] != DISCOVERY_TOPIC_PREFIX.strip("/"):
return None
if parts[-1] != "config":
return None
return {
"component": parts[1],
"node_id": "/".join(parts[2:-1]),
}
async def handle_discovery_message(
topic: str, payload: str, db: AsyncSession, mqtt_client: MqttClient
) -> Device | None:
"""Handle HA Discovery config message, auto-register device and subscribe to state topic"""
parsed = parse_discovery_topic(topic)
if not parsed:
return None
try:
config = json.loads(payload)
except json.JSONDecodeError:
logger.warning("Invalid JSON in discovery payload for %s", topic)
return None
state_topic = config.get("state_topic")
command_topic = config.get("command_topic")
device_name = config.get("name", config.get("device", {}).get("name", parsed["node_id"]))
device_class = config.get("device_class", "")
if not state_topic:
logger.warning("Discovery config for %s has no state_topic", topic)
return None
# Check if already exists (deduplicate by discovery_topic)
from sqlalchemy import select
result = await db.execute(
select(Device).where(Device.discovery_topic == topic)
)
existing = result.scalar_one_or_none()
if existing:
logger.debug("Device already registered: %s", topic)
return existing
from mqtt_home.schemas import DeviceCreate
device = await create_device(db, DeviceCreate(
name=device_name,
type=parsed["component"],
protocol="ha_discovery",
mqtt_topic=state_topic,
command_topic=command_topic,
))
# Update discovery-related fields
device.discovery_topic = topic
device.discovery_payload = payload
device.attributes = json.dumps({
"device_class": device_class,
"unit_of_measurement": config.get("unit_of_measurement"),
"icon": config.get("icon"),
})
await db.commit()
await db.refresh(device)
# Subscribe to state topic
await mqtt_client.subscribe(state_topic, qos=1)
logger.info("HA Discovery device registered: %s -> %s", device.name, state_topic)
return device

17
src/mqtt_home/emqx_api.py
View File

@@ -18,12 +18,25 @@ class EmqxApiClient:
async def get_broker_status(self) -> dict[str, Any]: async def get_broker_status(self) -> dict[str, Any]:
resp = await self._client.get("/status") resp = await self._client.get("/status")
resp.raise_for_status() resp.raise_for_status()
return resp.json().get("data", {}) # /status may return plain text, try JSON first
try:
data = resp.json()
if isinstance(data, dict) and "data" in data:
return data["data"]
return data if isinstance(data, dict) else {"raw": resp.text}
except Exception:
return {"raw": resp.text}
async def get_metrics(self) -> dict[str, Any]: async def get_metrics(self) -> dict[str, Any]:
resp = await self._client.get("/metrics") resp = await self._client.get("/metrics")
resp.raise_for_status() resp.raise_for_status()
return resp.json().get("data", {}) data = resp.json()
# /metrics returns a list in some EMQX versions
if isinstance(data, list) and data:
return data[0] if isinstance(data[0], dict) else {}
if isinstance(data, dict) and "data" in data:
return data["data"] if isinstance(data["data"], dict) else {}
return data if isinstance(data, dict) else {}
async def get_clients(self, limit: int = 100) -> list[dict[str, Any]]: async def get_clients(self, limit: int = 100) -> list[dict[str, Any]]:
resp = await self._client.get("/clients", params={"limit": limit}) resp = await self._client.get("/clients", params={"limit": limit})

153
src/mqtt_home/main.py Normal file
View File

@@ -0,0 +1,153 @@
import asyncio
import logging
import sys
import pathlib
from contextlib import asynccontextmanager
# Windows compatibility: use SelectorEventLoop for paho-mqtt (add_reader/add_writer)
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from mqtt_home.config import get_settings
from mqtt_home.database import init_db, get_session_factory, Base
from mqtt_home.mqtt_client import MqttClient
from mqtt_home.emqx_api import EmqxApiClient
from mqtt_home.discovery import handle_discovery_message
from mqtt_home.device_registry import handle_state_update
from mqtt_home.broker_devices import BrokerDeviceRegistry
from mqtt_home.rule_registry import list_rules
from mqtt_home.api import api_router
from mqtt_home.ws import websocket_endpoint, broadcast_device_update
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger(__name__)
# Default topic rules created on first startup when no rules exist
DEFAULT_RULES = [
{
"name": "通用设备",
"topic_pattern": "home/+",
"device_type": "switch",
"command_template": "home/{device_id}/set",
"state_value_path": None,
},
]
@asynccontextmanager
async def lifespan(app: FastAPI):
settings = get_settings()
await init_db()
logger.info("Database initialized")
emqx = EmqxApiClient(settings)
app.state.emqx_client = emqx
logger.info("EMQX API client initialized")
mqtt = MqttClient(settings)
app.state.mqtt_client = mqtt
session_factory = get_session_factory()
# Create default rules if none exist (first startup)
async with session_factory() as db:
from mqtt_home.rule_registry import list_rules, create_rule
existing = await list_rules(db)
if not existing:
from mqtt_home.schemas import RuleCreate
for default in DEFAULT_RULES:
await create_rule(db, RuleCreate(**default))
logger.info("Created default rule: %s", default["name"])
async def on_discovery(topic: str, payload: str):
async with session_factory() as db:
await handle_discovery_message(topic, payload, db, mqtt)
async def on_state(topic: str, payload: str):
# Original: update DB devices
async with session_factory() as db:
device = await handle_state_update(db, topic, payload)
if device:
await broadcast_device_update(device.id, {
"state": device.state,
"is_online": device.is_online,
"last_seen": device.last_seen.isoformat() if device.last_seen else None,
})
# New: match topic format rules, update broker device registry
rules = getattr(app.state, "topic_rules", [])
registry: BrokerDeviceRegistry = getattr(app.state, "broker_registry", None)
if registry and rules:
for rule in rules:
if not rule.is_enabled:
continue
updated = registry.update_or_create(topic, payload, rule)
if updated:
await broadcast_device_update(updated.id, {
"state": updated.state,
"is_online": updated.is_online,
"last_seen": updated.last_seen.isoformat() if updated.last_seen else None,
"source": "broker",
})
mqtt.on_message("homeassistant/#", on_discovery)
mqtt.on_message("home/#", on_state)
await mqtt.start()
logger.info("MQTT client started")
# Initialize broker device registry
broker_registry = BrokerDeviceRegistry()
app.state.broker_registry = broker_registry
# Load topic rules from DB
async with session_factory() as db:
app.state.topic_rules = [r for r in await list_rules(db) if r.is_enabled]
logger.info("Loaded %d topic format rules", len(app.state.topic_rules))
yield
await mqtt.stop()
await emqx.close()
logger.info("Shutdown complete")
app = FastAPI(title="MQTT Home", version="0.1.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(api_router)
app.websocket("/ws/devices")(websocket_endpoint)
@app.get("/health")
async def health():
mqtt = getattr(app.state, "mqtt_client", None)
return {
"status": "ok",
"mqtt_connected": mqtt.is_connected if mqtt else False,
}
# Serve frontend static files in production
_frontend_dist = pathlib.Path(__file__).parent.parent.parent / "frontend" / "dist"
if _frontend_dist.exists():
app.mount("/assets", StaticFiles(directory=str(_frontend_dist / "assets")), name="assets")
@app.get("/{full_path:path}")
async def serve_spa(full_path: str):
file = _frontend_dist / full_path
if file.exists() and file.is_file():
return FileResponse(str(file))
return FileResponse(str(_frontend_dist / "index.html"))

15
src/mqtt_home/models.py
View File

@@ -46,4 +46,17 @@ class DeviceLog(Base):
payload = Column(Text, nullable=False) payload = Column(Text, nullable=False)
timestamp = Column(DateTime(timezone=True), nullable=False, default=_utcnow) timestamp = Column(DateTime(timezone=True), nullable=False, default=_utcnow)
device = relationship("Device", back_populates="logs") device = relationship("Device", back_populates="logs")
class TopicFormatRule(Base):
__tablename__ = "topic_format_rules"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(200), nullable=False)
topic_pattern = Column(String(500), nullable=False) # MQTT pattern: "home/+/state"
device_type = Column(String(50), nullable=False, default="switch")
command_template = Column(String(500), nullable=True) # "home/{device_id}/set"
state_value_path = Column(String(200), nullable=True) # "state"
is_enabled = Column(Boolean, nullable=False, default=True)
created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow)

57
src/mqtt_home/rule_registry.py Normal file
View File

@@ -0,0 +1,57 @@
import logging
from typing import Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from mqtt_home.models import TopicFormatRule
from mqtt_home.schemas import RuleCreate, RuleUpdate
logger = logging.getLogger(__name__)
async def list_rules(db: AsyncSession) -> list[TopicFormatRule]:
result = await db.execute(select(TopicFormatRule).order_by(TopicFormatRule.created_at.desc()))
return list(result.scalars().all())
async def get_rule(db: AsyncSession, rule_id: int) -> Optional[TopicFormatRule]:
return await db.get(TopicFormatRule, rule_id)
async def create_rule(db: AsyncSession, data: RuleCreate) -> TopicFormatRule:
rule = TopicFormatRule(
name=data.name,
topic_pattern=data.topic_pattern,
device_type=data.device_type,
command_template=data.command_template,
state_value_path=data.state_value_path,
is_enabled=data.is_enabled,
)
db.add(rule)
await db.commit()
await db.refresh(rule)
logger.info("Rule created: %s (id=%d, pattern=%s)", rule.name, rule.id, rule.topic_pattern)
return rule
async def update_rule(db: AsyncSession, rule_id: int, data: RuleUpdate) -> Optional[TopicFormatRule]:
rule = await db.get(TopicFormatRule, rule_id)
if not rule:
return None
update_data = data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(rule, key, value)
await db.commit()
await db.refresh(rule)
return rule
async def delete_rule(db: AsyncSession, rule_id: int) -> bool:
rule = await db.get(TopicFormatRule, rule_id)
if not rule:
return False
await db.delete(rule)
await db.commit()
logger.info("Rule deleted: id=%d", rule_id)
return True

47
src/mqtt_home/schemas.py
View File

@@ -63,8 +63,55 @@ class BrokerTopic(BaseModel):
node: Optional[str] = None node: Optional[str] = None
class BrokerDeviceResponse(BaseModel):
id: str
name: str
type: str
protocol: str = "topic_rule"
mqtt_topic: str
command_topic: str | None = None
state: str | None = None
is_online: bool = False
last_seen: datetime | None = None
rule_id: int = 0
class DashboardStats(BaseModel): class DashboardStats(BaseModel):
total_devices: int total_devices: int
online_devices: int online_devices: int
offline_devices: int offline_devices: int
recent_logs: list[DeviceLogResponse] recent_logs: list[DeviceLogResponse]
broker_topics: list[str] = []
broker_device_count: int = 0
broker_online_count: int = 0
mqtt_connected: bool = False
class RuleCreate(BaseModel):
name: str
topic_pattern: str
device_type: str = "switch"
command_template: str | None = None
state_value_path: str | None = None
is_enabled: bool = True
class RuleUpdate(BaseModel):
name: str | None = None
topic_pattern: str | None = None
device_type: str | None = None
command_template: str | None = None
state_value_path: str | None = None
is_enabled: bool | None = None
class RuleResponse(BaseModel):
id: int
name: str
topic_pattern: str
device_type: str
command_template: str | None = None
state_value_path: str | None = None
is_enabled: bool
created_at: datetime
model_config = {"from_attributes": True}

131
src/mqtt_home/topic_matcher.py Normal file
View File

@@ -0,0 +1,131 @@
import json
from typing import Optional
def match_topic(topic: str, pattern: str) -> dict[str, str] | None:
"""Match an MQTT topic against a pattern with + and # wildcards.
Converts MQTT pattern to regex: + -> ([^/]+), # -> (.+)
Returns dict of capture groups indexed by wildcard position, e.g. {"1": "fire"}
Returns None if no match.
"""
topic_parts = topic.split("/")
pattern_parts = pattern.split("/")
captures: dict[str, str] = {}
t_idx = 0
p_idx = 0
while t_idx < len(topic_parts) and p_idx < len(pattern_parts):
pat = pattern_parts[p_idx]
if pat == "#":
# # matches everything remaining (must be last in pattern per MQTT spec)
captures[str(p_idx)] = "/".join(topic_parts[t_idx:])
t_idx = len(topic_parts)
p_idx += 1
elif pat == "+":
# + matches exactly one level
captures[str(p_idx)] = topic_parts[t_idx]
t_idx += 1
p_idx += 1
elif pat == topic_parts[t_idx]:
t_idx += 1
p_idx += 1
else:
return None
# After main loop, check remaining parts
if p_idx < len(pattern_parts):
# Remaining pattern parts must all be empty-string compatible or #
remaining_pattern = pattern_parts[p_idx:]
if remaining_pattern == ["#"] or remaining_pattern == []:
if remaining_pattern == ["#"]:
captures[str(p_idx)] = ""
p_idx = len(pattern_parts)
else:
return None
if t_idx < len(topic_parts):
return None
if captures:
return captures
return None
def extract_device_id(topic: str, pattern: str) -> str | None:
"""Extract the device identifier from a topic based on pattern.
For patterns with + wildcards, returns the last + match.
For patterns with #, returns the part matched by # (last segment).
Returns None if topic doesn't match pattern.
Examples:
extract_device_id("home/fire/state", "home/+/state") -> "fire"
extract_device_id("home/fire/brightness", "home/fire/#") -> "brightness"
"""
result = match_topic(topic, pattern)
if result is None:
return None
# Check if # was used: # always captures everything remaining
# We need to check the pattern for # to identify its index
pattern_parts = pattern.split("/")
if "#" in pattern_parts:
hash_idx = str(pattern_parts.index("#"))
if hash_idx in result:
value = result[hash_idx]
# For #, return the last segment
if value:
return value.split("/")[-1]
return ""
# For + wildcards, return the last + match (highest index)
plus_indices = [
str(i) for i, p in enumerate(pattern_parts) if p == "+"
]
if plus_indices:
last_plus_idx = plus_indices[-1]
return result[last_plus_idx]
return None
def build_command_topic(command_template: str | None, device_id: str) -> str | None:
"""Replace {device_id} in command template with actual value.
Example: build_command_topic("home/{device_id}/set", "fire") -> "home/fire/set"
Returns None if template is None or empty.
"""
if not command_template:
return None
return command_template.replace("{device_id}", device_id)
def extract_state_value(payload: str, state_value_path: str | None) -> str:
"""Extract state value from JSON payload.
If state_value_path is set, extract that key from JSON payload.
If JSON parsing fails or path not found, return original payload.
If state_value_path is None, return entire payload string.
Examples:
extract_state_value('{"state":"on"}', "state") -> "on"
extract_state_value('{"state":"on"}', None) -> '{"state":"on"}'
extract_state_value("plain text", "state") -> "plain text"
extract_state_value('{"brightness":255}', "brightness") -> "255"
"""
if state_value_path is None:
return payload
try:
data = json.loads(payload)
if not isinstance(data, dict):
return payload
value = data.get(state_value_path)
if value is None:
return payload
return str(value)
except (json.JSONDecodeError, TypeError, ValueError):
return payload

50
src/mqtt_home/ws.py Normal file
View File

@@ -0,0 +1,50 @@
import asyncio
import json
import logging
from fastapi import WebSocket, WebSocketDisconnect
logger = logging.getLogger(__name__)
class ConnectionManager:
def __init__(self):
self._connections: list[WebSocket] = []
async def connect(self, ws: WebSocket):
await ws.accept()
self._connections.append(ws)
logger.info("WebSocket connected, total: %d", len(self._connections))
def disconnect(self, ws: WebSocket):
self._connections.remove(ws)
logger.info("WebSocket disconnected, total: %d", len(self._connections))
async def broadcast(self, message: dict):
dead = []
for ws in self._connections:
try:
await ws.send_json(message)
except Exception:
dead.append(ws)
for ws in dead:
self.disconnect(ws)
ws_manager = ConnectionManager()
async def websocket_endpoint(ws: WebSocket):
await ws_manager.connect(ws)
try:
while True:
await ws.receive_text()
except WebSocketDisconnect:
ws_manager.disconnect(ws)
async def broadcast_device_update(device_id: str, data: dict):
await ws_manager.broadcast({
"type": "device_update",
"device_id": device_id,
**data,
})

29
tests/test_api_broker.py Normal file
View File

@@ -0,0 +1,29 @@
import pytest
from httpx import AsyncClient, ASGITransport
from mqtt_home.main import app
@pytest.fixture
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
async def test_broker_status_no_client(client):
resp = await client.get("/api/broker/status")
assert resp.status_code == 503
async def test_broker_clients_no_client(client):
resp = await client.get("/api/broker/clients")
assert resp.status_code == 503
async def test_health_endpoint(client):
resp = await client.get("/health")
assert resp.status_code == 200
data = resp.json()
assert "status" in data
assert "mqtt_connected" in data

93
tests/test_api_devices.py Normal file
View File

@@ -0,0 +1,93 @@
import pytest
from httpx import AsyncClient, ASGITransport
from mqtt_home.main import app
from mqtt_home.database import Base, get_engine
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.pool import StaticPool
from mqtt_home.database import get_db
TEST_DB = "sqlite+aiosqlite:///:memory:"
@pytest.fixture
async def client():
engine = create_async_engine(TEST_DB, connect_args={"check_same_thread": False}, poolclass=StaticPool)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
test_factory = async_sessionmaker(engine, expire_on_commit=False)
async def override_get_db():
async with test_factory() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
app.dependency_overrides.clear()
await engine.dispose()
async def test_get_devices_empty(client):
resp = await client.get("/api/devices")
assert resp.status_code == 200
assert resp.json() == []
async def test_create_device(client):
resp = await client.post("/api/devices", json={
"name": "客厅灯",
"type": "light",
"mqtt_topic": "home/light",
"command_topic": "home/light/set",
})
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "客厅灯"
assert data["type"] == "light"
async def test_get_device_detail(client):
create_resp = await client.post("/api/devices", json={
"name": "灯", "type": "switch", "mqtt_topic": "t"
})
device_id = create_resp.json()["id"]
resp = await client.get(f"/api/devices/{device_id}")
assert resp.status_code == 200
assert resp.json()["name"] == "灯"
async def test_get_device_not_found(client):
resp = await client.get("/api/devices/nonexistent")
assert resp.status_code == 404
async def test_update_device(client):
create_resp = await client.post("/api/devices", json={
"name": "灯", "type": "switch", "mqtt_topic": "t"
})
device_id = create_resp.json()["id"]
resp = await client.put(f"/api/devices/{device_id}", json={"name": "新名字"})
assert resp.status_code == 200
assert resp.json()["name"] == "新名字"
async def test_delete_device(client):
create_resp = await client.post("/api/devices", json={
"name": "灯", "type": "switch", "mqtt_topic": "t"
})
device_id = create_resp.json()["id"]
resp = await client.delete(f"/api/devices/{device_id}")
assert resp.status_code == 204
async def test_delete_device_not_found(client):
resp = await client.delete("/api/devices/nonexistent")
assert resp.status_code == 404

86
tests/test_broker_devices.py Normal file
View File

@@ -0,0 +1,86 @@
from types import SimpleNamespace
from mqtt_home.broker_devices import BrokerDevice, BrokerDeviceRegistry
def make_rule(id=1, topic_pattern="home/+/state", device_type="switch",
command_template="home/{device_id}/set", state_value_path="state",
is_enabled=True):
return SimpleNamespace(
id=id, topic_pattern=topic_pattern, device_type=device_type,
command_template=command_template, state_value_path=state_value_path,
is_enabled=is_enabled,
)
def test_create_device_on_match():
registry = BrokerDeviceRegistry()
rule = make_rule()
device = registry.update_or_create("home/fire/state", '{"state":"on"}', rule)
assert device is not None
assert device.id == "broker:fire"
assert device.name == "fire"
assert device.type == "switch"
assert device.state == "on"
assert device.command_topic == "home/fire/set"
assert device.is_online is True
assert device.rule_id == 1
def test_update_existing_device():
registry = BrokerDeviceRegistry()
rule = make_rule()
registry.update_or_create("home/fire/state", '{"state":"on"}', rule)
device = registry.update_or_create("home/fire/state", '{"state":"off"}', rule)
assert device.state == "off"
assert len(registry.get_all()) == 1
def test_different_topics_different_devices():
registry = BrokerDeviceRegistry()
rule = make_rule()
registry.update_or_create("home/fire/state", '{"state":"on"}', rule)
registry.update_or_create("home/servo/state", '{"state":"off"}', rule)
assert len(registry.get_all()) == 2
def test_no_match_returns_none():
registry = BrokerDeviceRegistry()
rule = make_rule(topic_pattern="home/+/state")
result = registry.update_or_create("living/light/status", '{"state":"on"}', rule)
assert result is None
def test_remove_by_rule():
registry = BrokerDeviceRegistry()
rule1 = make_rule(id=1)
rule2 = make_rule(id=2, topic_pattern="sensor/+/data")
registry.update_or_create("home/fire/state", '{"state":"on"}', rule1)
registry.update_or_create("sensor/temp/data", '{"value":25}', rule2)
assert len(registry.get_all()) == 2
removed = registry.remove_by_rule(1)
assert removed == ["broker:fire"]
assert len(registry.get_all()) == 1
assert registry.get("broker:fire") is None
def test_full_payload_as_state():
registry = BrokerDeviceRegistry()
rule = make_rule(state_value_path=None)
device = registry.update_or_create("home/fire/state", '{"state":"on","brightness":128}', rule)
assert device.state == '{"state":"on","brightness":128}'
def test_get_device():
registry = BrokerDeviceRegistry()
rule = make_rule()
registry.update_or_create("home/fire/state", '{"state":"on"}', rule)
assert registry.get("broker:fire") is not None
assert registry.get("broker:nonexistent") is None
def test_clear():
registry = BrokerDeviceRegistry()
rule = make_rule()
registry.update_or_create("home/fire/state", '{"state":"on"}', rule)
registry.clear()
assert len(registry.get_all()) == 0

27
tests/test_cli.py Normal file
View File

@@ -0,0 +1,27 @@
import pytest
from click.testing import CliRunner
from mqtt_home.cli import cli
def test_device_list_empty(monkeypatch):
runner = CliRunner()
async def mock_init():
pass
async def mock_list(db):
return []
monkeypatch.setattr("mqtt_home.cli.init_db", mock_init)
monkeypatch.setattr("mqtt_home.cli.list_devices", mock_list)
result = runner.invoke(cli, ["device", "list"])
assert result.exit_code == 0
assert "暂无设备" in result.output
def test_cli_groups():
runner = CliRunner()
result = runner.invoke(cli, ["--help"])
assert "device" in result.output
assert "broker" in result.output

2
tests/test_config.py
View File

@@ -3,7 +3,7 @@ from mqtt_home.config import Settings
def test_default_settings(): def test_default_settings():
s = Settings() s = Settings(_env_file=None)
assert s.mqtt_host == "localhost" assert s.mqtt_host == "localhost"
assert s.mqtt_port == 1883 assert s.mqtt_port == 1883
assert s.web_port == 8000 assert s.web_port == 8000

78
tests/test_discovery.py Normal file
View File

@@ -0,0 +1,78 @@
import json
import pytest
from unittest.mock import AsyncMock
from mqtt_home.discovery import parse_discovery_topic, handle_discovery_message
from mqtt_home.config import Settings
from mqtt_home.mqtt_client import MqttClient
@pytest.fixture
def settings():
return Settings(
mqtt_host="localhost",
emqx_api_url="http://localhost:18083/api/v5",
emqx_api_key="test-key",
emqx_api_secret="test-secret",
database_url="sqlite+aiosqlite:///:memory:",
)
def test_parse_discovery_topic_valid():
result = parse_discovery_topic("homeassistant/light/abc123/config")
assert result is not None
assert result["component"] == "light"
assert result["node_id"] == "abc123"
def test_parse_discovery_topic_nested_node():
result = parse_discovery_topic("homeassistant/sensor/room/temperature/config")
assert result is not None
assert result["component"] == "sensor"
assert result["node_id"] == "room/temperature"
def test_parse_discovery_topic_invalid():
assert parse_discovery_topic("other/topic/config") is None
assert parse_discovery_topic("homeassistant/light/abc") is None
assert parse_discovery_topic("homeassistant/light/abc/status") is None
async def test_handle_discovery_creates_device(db_session, settings):
mqtt_client = MqttClient(settings)
mqtt_client.subscribe = AsyncMock()
payload = json.dumps({
"name": "客厅灯",
"state_topic": "home/living/light/status",
"command_topic": "home/living/light/set",
"device_class": "light",
})
device = await handle_discovery_message(
"homeassistant/light/living_room/config",
payload,
db_session,
mqtt_client,
)
assert device is not None
assert device.name == "客厅灯"
assert device.protocol == "ha_discovery"
assert device.mqtt_topic == "home/living/light/status"
mqtt_client.subscribe.assert_called_once_with("home/living/light/status", qos=1)
async def test_handle_discovery_duplicate(db_session, settings):
mqtt_client = MqttClient(settings)
mqtt_client.subscribe = AsyncMock()
payload = json.dumps({
"name": "灯",
"state_topic": "home/light",
})
await handle_discovery_message("homeassistant/light/test/config", payload, db_session, mqtt_client)
device2 = await handle_discovery_message("homeassistant/light/test/config", payload, db_session, mqtt_client)
# Same discovery_topic should not create duplicate
assert device2 is not None
mqtt_client.subscribe.assert_called_once() # Only called once

66
tests/test_rule_registry.py Normal file
View File

@@ -0,0 +1,66 @@
import pytest
from mqtt_home.rule_registry import list_rules, get_rule, create_rule, update_rule, delete_rule
from mqtt_home.schemas import RuleCreate, RuleUpdate
@pytest.mark.asyncio
async def test_create_rule(db_session):
rule = await create_rule(db_session, RuleCreate(
name="Test Rule",
topic_pattern="home/+/state",
device_type="switch",
command_template="home/{device_id}/set",
state_value_path="state",
))
assert rule.id is not None
assert rule.name == "Test Rule"
assert rule.topic_pattern == "home/+/state"
assert rule.is_enabled is True
@pytest.mark.asyncio
async def test_list_rules(db_session):
await create_rule(db_session, RuleCreate(name="Rule 1", topic_pattern="a/+"))
await create_rule(db_session, RuleCreate(name="Rule 2", topic_pattern="b/+"))
rules = await list_rules(db_session)
assert len(rules) == 2
@pytest.mark.asyncio
async def test_get_rule(db_session):
created = await create_rule(db_session, RuleCreate(name="Test", topic_pattern="x/+"))
rule = await get_rule(db_session, created.id)
assert rule is not None
assert rule.name == "Test"
@pytest.mark.asyncio
async def test_get_rule_not_found(db_session):
rule = await get_rule(db_session, 99999)
assert rule is None
@pytest.mark.asyncio
async def test_update_rule(db_session):
created = await create_rule(db_session, RuleCreate(name="Original", topic_pattern="x/+"))
updated = await update_rule(db_session, created.id, RuleUpdate(name="Updated"))
assert updated.name == "Updated"
assert updated.topic_pattern == "x/+" # unchanged
@pytest.mark.asyncio
async def test_update_rule_not_found(db_session):
result = await update_rule(db_session, 99999, RuleUpdate(name="X"))
assert result is None
@pytest.mark.asyncio
async def test_delete_rule(db_session):
created = await create_rule(db_session, RuleCreate(name="To Delete", topic_pattern="x/+"))
assert await delete_rule(db_session, created.id) is True
assert await get_rule(db_session, created.id) is None
@pytest.mark.asyncio
async def test_delete_rule_not_found(db_session):
assert await delete_rule(db_session, 99999) is False

74
tests/test_topic_matcher.py Normal file
View File

@@ -0,0 +1,74 @@
from mqtt_home.topic_matcher import match_topic, extract_device_id, build_command_topic, extract_state_value
def test_match_topic_single_plus():
assert match_topic("home/fire/state", "home/+/state") == {"1": "fire"}
def test_match_topic_no_match():
assert match_topic("home/fire/state", "home/living/state") is None
def test_match_topic_hash():
# # at end matches remaining segments
result = match_topic("home/fire/brightness", "home/fire/#")
assert result is not None
assert "2" in result # # is at index 2
def test_match_topic_multiple_plus():
assert match_topic("home/fire/living/state", "home/+/+/state") == {"1": "fire", "2": "living"}
def test_match_topic_hash_matches_empty():
assert match_topic("home/fire/", "home/fire/#") is not None
def test_extract_device_id_single_plus():
assert extract_device_id("home/fire/state", "home/+/state") == "fire"
def test_extract_device_id_no_match():
assert extract_device_id("home/fire/state", "home/living/state") is None
def test_extract_device_id_hash():
assert extract_device_id("home/fire/brightness", "home/fire/#") == "brightness"
def test_extract_device_id_multiple_plus():
# Returns last + match
assert extract_device_id("home/fire/living/state", "home/+/+/state") == "living"
def test_build_command_topic_basic():
assert build_command_topic("home/{device_id}/set", "fire") == "home/fire/set"
def test_build_command_topic_none():
assert build_command_topic(None, "fire") is None
assert build_command_topic("", "fire") is None
def test_extract_state_value_json_path():
assert extract_state_value('{"state":"on"}', "state") == "on"
def test_extract_state_value_no_path():
assert extract_state_value('{"state":"on"}', None) == '{"state":"on"}'
def test_extract_state_value_plain_text():
assert extract_state_value("plain text", "state") == "plain text"
def test_extract_state_value_nested_path():
assert extract_state_value('{"brightness":255}', "brightness") == "255"
def test_extract_state_value_missing_key():
assert extract_state_value('{"temperature":22}', "humidity") == '{"temperature":22}'
def test_extract_state_value_numeric_value():
assert extract_state_value('{"brightness":255}', "brightness") == "255"