基于 KaiwuDB 与 CodeArts 智能体的智能家居本地化数据处理方案

系统采用分层架构设计，所有模块均基于本地运行，数据全量存储在 KaiwuDB。项目结构包含设备接入层、数据处理层、规则引擎层及存储层，核心目录如下：

smarthome-system/
├── device-adapter/      # 协议适配器
├── data-processor/      # 数据处理
├── rule-engine/         # 规则引擎
├── storage/             # 存储策略
└── web-api/             # Web API 服务

数据库表设计

时序表用于存储高频采集的设备数据，关系表用于存储设备信息与规则配置。

-- 创建时序数据库
CREATE TS DATABASE IF NOT EXISTS tsdb;
USE tsdb;

-- 传感器时序数据表
CREATE TABLE IF NOT EXISTS home_sensor (
    ts TIMESTAMP NOT NULL,
    value FLOAT,
    status VARCHAR(32),
    file_path VARCHAR(128)
) ATTRIBUTES (
    device_id VARCHAR(64) NOT NULL,
    data_type VARCHAR(32) NOT NULL,
    location VARCHAR(32),
    device_model VARCHAR(64)
) PRIMARY TAGS(device_id) RETENTIONS 20D ACTIVETIME 12h;

-- 设备信息关系表
USE defaultdb;
CREATE TABLE IF NOT EXISTS devices (
    id INT PRIMARY KEY,
    name STRING NOT NULL,
    device_id STRING NOT NULL UNIQUE,
    device_type STRING NOT NULL,
    protocol STRING NOT NULL,
    location STRING,
    status STRING DEFAULT 'online',
    config JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

核心代码实现

Web API 服务基于 Flask 构建，提供数据接收与控制接口。主程序需模拟数据库连接或直接调用 Docker 容器内的 SQL 客户端。

#!/usr/bin/env python3
import sys
import os
from flask import Flask, jsonify, request

app = Flask(__name__)

# 模拟数据库连接类
class MockDB:
    def execute_query(self, query, params=None):
        # 实际场景中应替换为真实的 KaiwuDB 驱动连接
        # 此处演示通过 subprocess 调用容器内 SQL 客户端
        import subprocess
        try:
            cmd = ['docker', 'exec', 'kwdb-server', '/kaiwudb/bin/kwbase', 'sql', '--insecure', '--execute', query]
            result = subprocess.run(cmd, capture_output=True, text=True)
            if result.returncode == 0:
                lines = result.stdout.strip().split('\n')
                if len(lines) > 1:
                    headers = lines[0].split('\t')
                    data = []
                    for line in lines[1:]:
                        values = line.split('\t')
                        if len(values) == len(headers):
                            data.append(values)
                    return data
        except Exception as e:
            print(f"查询失败：{e}")
        return []

db = MockDB()

@app.route('/api/devices')
def get_devices():
    query = "SELECT id, name, device_id, device_type FROM devices ORDER BY id"
    results = db.execute_query(query)
    devices = []
    if results and len(results) > 1:
        headers = results[0]
        for row in results[1:]:
            devices.append(dict(zip(headers, row)))
    return jsonify(devices)

@app.route('/api/data/receive', methods=['POST'])
def receive_data():
    try:
        data = request.json
        if not data or 'device_id' not in data:
            return jsonify({'error': '缺少必要字段'}), 400
        
        device_id = data['device_id']
        value = data.get('value', 0)
        data_type = data.get('data_type', 'value')
        
        # 注意：生产环境请使用参数化查询防止注入
        query = f"INSERT INTO sensor_data (time, device_id, data_type, value) VALUES (NOW(), '{device_id}', '{data_type}', {value})"
        db.execute_query(query)
        return jsonify({'success': True})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5001, debug=True)

系统运行与测试

启动 Web API 服务后，可通过浏览器访问界面查看设备列表、实时数据及自动化规则状态。使用 curl 命令测试接口：

# 上报传感器数据
curl -X POST http://localhost:5001/api/data/receive \
-H "Content-Type: application/json" \
-d '{"device_id": "sensor_temp_livingroom", "value": 26.5, "data_type": "temperature"}'

# 获取设备列表
curl http://localhost:5001/api/devices

测试通过后，系统将能够持续接收本地设备数据，并根据预设规则触发联动动作，全程无需云端参与。

总结

本方案以 KaiwuDB 为数据基座，CodeArts 代码智能体为自动化开发引擎，实现了智能家居本地化数据处理系统的快速构建。通过自然语言驱动，降低了技术门槛，新手也可按步骤复现。随着 AIoT 技术发展，本地化数据处理将成为主流趋势，KaiwuDB 多模数据库让开发者能实现轻量、高效、安全的智能场景落地。