System Operational
I build things with code, almost always with an AI at my side. It feels less like giving commands and more like having a conversation—asking questions, trying ideas, and seeing what unfolds.
My approach is rooted in minimalism and simplicity, where less is often more. I believe in creating designs that are not only beautiful but also intuitive and user-friendly.
When I'm not working, you can find me photographing, reading design books, or enjoying the outdoors.
RISE.L.net 是由 RISE.L 课题组开发的轻量级、高可扩展的边缘设备通信标准与库。该项目源于课题组在智能安全监测系统（如头盔检测、安全帽识别等）实际应用中的经验总结，旨在为嵌入式 AI 设备与云端服务器之间的通信提供一套简单、可靠、易扩展的解决方案。
传统的边缘设备接入云端通常需要编写大量的网络通信、设备注册、心跳维护等代码。RISE.L.net 将这些复杂逻辑封装为简洁的 API，使得设备接入只需 3 行代码。
# 服务器端
from riselnet_server import quick_server
quick_server(8080, lambda device_id, data: print(f"设备 {device_id} 上报: {data}"))
# 客户端（MicroPython/嵌入式设备）
from riselnet import quick_start
device = quick_start("http://server:8080", "WiFi_SSID", "password")
device.report("detection", {"helmet": 1, "no_helmet": 0})自动设备注册、周期性心跳机制（默认 30 秒）、离线超时检测（超过 90 秒）、实时数据上报（支持任意结构的 JSON 数据）
# 心跳机制实现（客户端）
def _heartbeat_loop(self):
"""心跳循环"""
while self._running:
self._send_heartbeat()
time.sleep(self.heartbeat_interval)
# 设备超时检测（服务器端）
def _check_device_timeout(self):
"""检查设备超时"""
while self._running:
devices = self._storage.get_all_devices(status='online')
for device in devices:
last_heartbeat_str = device.get('last_heartbeat')
if last_heartbeat_str:
last_heartbeat = datetime.strptime(last_heartbeat_str, '%Y-%m-%d %H:%M:%S')
if (datetime.now() - last_heartbeat).total_seconds() > self.device_timeout:
device_id = device['device_id']
self._storage.update_device_status(device_id, 'offline')
print(f"[RISE.L.net Server] 🔴 设备离线: {device_id}")
time.sleep(30)传输层抽象、中间件系统、插件系统、存储层抽象，支持多种协议和存储后端
# 核心抽象层
class Transport:
"""传输层抽象基类"""
def send(self, endpoint, data):
"""发送数据，返回 (success, response)"""
raise NotImplementedError
class Middleware:
"""中间件基类"""
def before_send(self, endpoint, data):
"""发送前处理，返回处理后的 data"""
return data
class Plugin:
"""插件基类"""
def on_load(self, server):
"""插件加载时调用"""
pass
class Storage:
"""存储层抽象基类"""
def save_device(self, device_id, data):
"""保存设备信息"""
raise NotImplementedError客户端：MicroPython（K230、ESP32、树莓派 Pico 等）；服务器端：Python 3.6+（Linux、Windows、macOS）
# 客户端高级用法
device = RISELDevice("http://server:8080")
device.use(RetryMiddleware(max_retries=3))
device.use(CompressionMiddleware())
device.transport(MQTTTransport("mqtt://server"))
device.start()
# 服务器端高级用法
server = RISELServer(port=8080)
server.use(AuthMiddleware(api_key="secret"))
server.use(RateLimitMiddleware(max_rps=100))
server.storage(RedisStorage("redis://localhost"))
server.route("/api/custom", custom_handler)
server.plugin(TelegramNotifier(token="..."))
server.start()头盔检测、安全帽识别等实时监测系统，支持边缘 AI 推理、云端管理、即时告警等功能
工业物联网设备管理、智能传感器网络、远程设备监控与维护、边缘计算数据采集
采用独立线程实现周期性心跳，默认间隔 30 秒，服务端通过时间戳比对检测设备离线状态（超过 90 秒判定为离线）。
# 心跳机制核心实现
def _heartbeat_loop(self):
"""心跳循环实现"""
while self._running:
# 发送心跳包，包含设备ID和时间戳
payload = {
'device_id': self.device_id,
'timestamp': datetime.now().isoformat()
}
success, response = self.transport.send('/heartbeat', payload)
if not success:
# 失败时进行重试
self._handle_retry()
time.sleep(self.heartbeat_interval)采用轻量级 JSON 序列化，支持数据压缩和批量上报，减少网络带宽占用。
# 数据压缩实现
class CompressionMiddleware(Middleware):
def before_send(self, endpoint, data):
# 对大数据进行压缩
if len(str(data)) > 1024:
compressed = self._compress(data)
return {'compressed': 1, 'data': compressed}
return data实现多级重试和离线缓存，当网络中断时自动缓存数据，网络恢复后重传。
# 离线缓存实现
def _cache_data(self, endpoint, data):
# 将数据缓存到本地文件
cache_key = f"{endpoint}_{int(time.time())}"
cache_data = {
'endpoint': endpoint,
'data': data,
'timestamp': time.time()
}
with open(f"./cache/{cache_key}.json", 'w') as f:
json.dump(cache_data, f)
def _replay_cache(self):
# 重传缓存的数据
for cache_file in os.listdir('./cache'):
with open(f"./cache/{cache_file}", 'r') as f:
cache_data = json.load(f)
success, _ = self.transport.send(
cache_data['endpoint'],
cache_data['data']
)
if success:
os.remove(f"./cache/{cache_file}")支持 API Key、JWT 令牌等多种认证方式，保护设备通信安全。
# 认证中间件实现
class AuthMiddleware(Middleware):
def __init__(self, api_key):
self.api_key = api_key
def before_send(self, endpoint, data):
# 添加认证信息
data['api_key'] = self.api_key
data['signature'] = self._generate_signature(data)
return data
def _generate_signature(self, data):
# 生成数据签名
sorted_data = sorted(data.items())
sign_str = ''.join([f"{k}={v}" for k, v in sorted_data])
return hashlib.sha256((sign_str + self.api_key).encode()).hexdigest()技术实现：在边缘设备（如 K230）上部署轻量级 AI 模型，实时检测头盔/安全帽佩戴情况，通过 RISE.L.net 将检测结果上报至云端。
核心功能： - 实时视频流处理（15fps） - 边缘 AI 推理（模型大小 < 5MB） - 云端数据聚合与分析 - 异常情况即时告警（短信/邮件通知）
应用价值：大幅降低工地安全事故率，减少人工巡检成本，提高管理效率。
技术实现：通过 RISE.L.net 连接工厂内的各类传感器和设备，实现统一管理和数据采集。
核心功能： - 设备状态实时监控 - 历史数据存储与分析 - 设备故障预测与预警 - 远程配置与升级
应用价值：提高设备利用率，减少停机时间，优化生产流程，降低运营成本。
技术实现：在园区或城市部署多节点传感器网络，通过 RISE.L.net 实现数据采集和协同工作。
核心功能： - 环境参数监测（温度、湿度、PM2.5等） - 传感器节点自组织网络 - 数据聚合与边缘计算 - 异常事件检测与上报
应用价值：为智慧城市建设提供数据支撑，改善环境质量，提升居民生活品质。
技术实现：通过 RISE.L.net 实现对分散在各地的设备进行远程监控和维护。
核心功能： - 设备运行状态实时监测 - 远程诊断与故障排除 - 固件远程升级 - 维护计划自动生成
应用价值：减少现场维护次数，降低维护成本，提高设备可靠性和使用寿命。
RISE.L (Recognition Intelligence Systems Edge Lab) 课题组专注于边缘智能 AI 系统和设备通信框架的研发，致力于将先进的人工智能技术应用于实际工业场景。
课题组核心项目是一套完整的工业级 AI 安全监控系统，基于 CanMV-K230 开发板实现边缘端实时 AI 推理，结合云端数据分析与管理，形成完整的安全监测闭环。