一、问题定义

在智慧养老领域，常见的技术问题是：设备报警灵敏，但信息流止步于监控大屏或手机App。当老人发生意外时，毫米波雷达触发警报，App向家属或社区中心发送通知，但最终的物理救助行动无法闭环。

这种“有预警、无处置”的模式，使得智能设备未能有效解决养老照护的“最后一公里”问题。本文从技术架构角度，分析如何通过设备、平台与线下服务的深度整合，构建完整的应急响应闭环。

---

二、系统架构设计

2.1 三层架构

text

┌─────────────────────────────────────────────────────────────┐
│                       服务层                                 │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐            │
│  │ 应急响应团队 │ │ 照护服务团队 │ │ 家属通知    │            │
│  └──────┬──────┘ └──────┬──────┘ └──────┬──────┘            │
├─────────┼───────────────┼───────────────┼────────────────────┤
│         │               │               │                    │
│         └───────────────┼───────────────┘                    │
│                         ▼                                    │
│                      平台层                                   │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐            │
│  │ AIoT接入网关 │ │ AI误报过滤  │ │ 调度中心     │            │
│  └──────┬──────┘ └──────┬──────┘ └──────┬──────┘            │
├─────────┼───────────────┼───────────────┼────────────────────┤
│         │               │               │                    │
│         └───────────────┼───────────────┘                    │
│                         ▼                                    │
│                      感知层                                   │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐            │
│  │ 毫米波雷达   │ │ SOS呼叫器   │ │ 智能床垫    │            │
│  └─────────────┘ └─────────────┘ └─────────────┘            │
└─────────────────────────────────────────────────────────────┘

2.2 数据流向与闭环

层级	输入	处理	输出
感知层	物理信号	点云滤波、姿态解算	结构化事件
平台层	边缘事件	AI二次研判、调度匹配	服务工单
服务层	工单	接单、到达、完成确认	闭环记录

---

三、感知层：毫米波雷达数据处理

3.1 点云数据预过滤

原始点云数据噪声多、数据量大，需在边缘侧进行预处理。

python

import numpy as np
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime

@dataclass
class PointCloud:
    """点云数据结构"""
    x: float   # x坐标(m)
    y: float   # y坐标(m)
    z: float   # z坐标(m)
    vx: float  # x方向速度(m/s)
    vy: float  # y方向速度(m/s)
    vz: float  # z方向速度(m/s)

@dataclass
class DetectedPerson:
    """检测到的人体"""
    centroid: Tuple[float, float, float]  # 质心坐标(x,y,z)
    velocity: Tuple[float, float, float]   # 速度(vx,vy,vz)
    height_ratio: float                    # 高度比(当前高度/站立高度)
    point_count: int                       # 包含的点云数量

class RadarEventProcessor:
    """毫米波雷达事件处理器"""
    
    def __init__(self, fall_velocity_threshold: float = 2.0):
        """
        fall_velocity_threshold: 跌倒判断的速度阈值(m/s)
        """
        self.fall_velocity_threshold = fall_velocity_threshold
        self.standing_height: Optional[float] = None
    
    def filter_static_noise(self, points: List[PointCloud]) -> List[PointCloud]:
        """静态杂波滤除"""
        dynamic_points = []
        for p in points:
            # 速度为0或接近0的为静态点，过滤掉
            if abs(p.vx) + abs(p.vy) + abs(p.vz) > 0.1:
                dynamic_points.append(p)
        return dynamic_points
    
    def cluster_points(self, points: List[PointCloud]) -> Optional[List[PointCloud]]:
        """
        点云聚类（简化版DBSCAN）
        将空间上接近的点归为一个目标
        """
        if len(points) < 3:
            return None
        
        # 简化实现：按空间距离聚类
        clusters = []
        remaining = points.copy()
        
        while remaining:
            seed = remaining.pop(0)
            cluster = [seed]
            i = 0
            while i < len(remaining):
                if self._distance(seed, remaining[i]) < 0.5:  # 0.5米内归为一类
                    cluster.append(remaining.pop(i))
                else:
                    i += 1
            clusters.append(cluster)
        
        # 返回最大的簇（最可能是人）
        if clusters:
            return max(clusters, key=len)
        return None
    
    def calculate_pose(self, cluster: List[PointCloud]) -> Optional[DetectedPerson]:
        """计算人体姿态"""
        if not cluster or len(cluster) < 3:
            return None
        
        # 计算质心
        centroid_x = np.mean([p.x for p in cluster])
        centroid_y = np.mean([p.y for p in cluster])
        centroid_z = np.mean([p.z for p in cluster])
        
        # 计算平均速度
        vel_x = np.mean([p.vx for p in cluster])
        vel_y = np.mean([p.vy for p in cluster])
        vel_z = np.mean([p.vz for p in cluster])
        
        # 计算高度范围
        z_values = [p.z for p in cluster]
        current_height = max(z_values) - min(z_values)
        
        # 初始化站立高度（取前几次检测的最大值）
        if self.standing_height is None and current_height > 1.2:
            self.standing_height = current_height
        
        height_ratio = current_height / self.standing_height if self.standing_height else 1.0
        
        return DetectedPerson(
            centroid=(centroid_x, centroid_y, centroid_z),
            velocity=(vel_x, vel_y, vel_z),
            height_ratio=height_ratio,
            point_count=len(cluster)
        )
    
    def detect_fall(self, person: DetectedPerson) -> Tuple[bool, float]:
        """
        跌倒检测
        返回 (是否跌倒, 置信度)
        """
        vx, vy, vz = person.velocity
        
        # 跌倒特征：垂直速度突变 + 高度骤降
        vertical_velocity = abs(vz)
        is_rapid_drop = vertical_velocity > self.fall_velocity_threshold
        is_low_posture = person.height_ratio < 0.4
        
        if is_rapid_drop and is_low_posture:
            # 置信度计算
            confidence = min(1.0, (vertical_velocity / self.fall_velocity_threshold) * 0.6 + (1 - person.height_ratio) * 0.4)
            return True, round(confidence, 2)
        
        return False, 0.0
    
    def process(self, raw_points: List[PointCloud]) -> Optional[Dict]:
        """处理一帧点云数据"""
        # 1. 静态杂波滤除
        dynamic_points = self.filter_static_noise(raw_points)
        
        if len(dynamic_points) < 3:
            return None
        
        # 2. 点云聚类
        cluster = self.cluster_points(dynamic_points)
        if not cluster:
            return None
        
        # 3. 姿态计算
        person = self.calculate_pose(cluster)
        if not person:
            return None
        
        # 4. 跌倒检测
        is_fall, confidence = self.detect_fall(person)
        
        return {
            'timestamp': datetime.now().isoformat(),
            'event_type': 'FALL_DETECTED' if is_fall else 'PRESENCE_DETECTED',
            'location': person.centroid,
            'confidence': confidence,
            'height_ratio': person.height_ratio,
            'point_count': person.point_count
        }
    
    def _distance(self, p1: PointCloud, p2: PointCloud) -> float:
        """计算两点间的欧氏距离"""
        return ((p1.x - p2.x)**2 + (p1.y - p2.y)**2 + (p1.z - p2.z)**2) ** 0.5


# 使用示例
processor = RadarEventProcessor(fall_velocity_threshold=2.0)

# 模拟正常活动点云
normal_points = [PointCloud(1.0, 2.0, 1.5, 0.1, 0.2, 0.0) for _ in range(20)]
result = processor.process(normal_points)
print("正常活动:", result)

# 模拟跌倒点云
fall_points = [PointCloud(1.0, 2.0, 0.3, 0.1, 0.2, -2.5) for _ in range(20)]
result = processor.process(fall_points)
print("跌倒事件:", result)

3.2 边缘事件上报格式

python

# 边缘侧上报的结构化事件
def format_edge_event(event_data: Dict, device_id: str) -> Dict:
    """格式化边缘事件"""
    return {
        'device_id': device_id,
        'event_type': event_data['event_type'],
        'timestamp': event_data['timestamp'],
        'data': {
            'location': event_data['location'],
            'confidence': event_data['confidence'],
            'height_ratio': event_data['height_ratio'],
            'point_count': event_data['point_count']
        },
        'edge_processed': True
    }

---

四、平台层：AI误报过滤与调度

4.1 云端AI误报过滤

python

from typing import Dict, List
import numpy as np

class CloudAIFilter:
    """云端AI误报过滤器"""
    
    def __init__(self, model_weights: Optional[Dict] = None):
        self.model_weights = model_weights or {
            'velocity_weight': 0.35,
            'height_ratio_weight': 0.35,
            'history_weight': 0.20,
            'confidence_weight': 0.10
        }
        self.event_history: Dict[str, List[Dict]] = {}
    
    def extract_features(self, edge_event: Dict, device_history: List[Dict]) -> np.ndarray:
        """特征提取"""
        # 提取事件特征
        velocity = abs(edge_event['data']['location'][2])  # Z轴速度绝对值
        height_ratio = edge_event['data']['height_ratio']
        confidence = edge_event['data']['confidence']
        
        # 计算历史误报率
        if device_history:
            false_positive_count = sum(1 for e in device_history if e.get('is_false_positive', False))
            history_fp_rate = false_positive_count / len(device_history)
        else:
            history_fp_rate = 0.0
        
        return np.array([velocity, height_ratio, confidence, history_fp_rate])
    
    def predict(self, features: np.ndarray) -> Tuple[bool, float]:
        """
        预测是否为真实事件
        返回 (是否为真实事件, 置信度)
        """
        # 加权评分（简化版逻辑回归）
        v, hr, conf, fp_rate = features
        
        # 真实事件的特征：高速度变化 + 低高度比 + 高置信度 + 低历史误报率
        score = (
            min(v / 3.0, 1.0) * self.model_weights['velocity_weight'] +
            (1 - hr) * self.model_weights['height_ratio_weight'] +
            conf * self.model_weights['confidence_weight'] +
            (1 - fp_rate) * self.model_weights['history_weight']
        )
        
        is_true_event = score > 0.6
        return is_true_event, round(score, 2)
    
    def evaluate_event(self, edge_event: Dict, device_history: List[Dict]) -> Tuple[bool, Dict]:
        """评估事件"""
        features = self.extract_features(edge_event, device_history)
        is_true, confidence = self.predict(features)
        
        return is_true, {
            'is_true_event': is_true,
            'confidence': confidence,
            'features': {
                'velocity': features[0],
                'height_ratio': features[1],
                'edge_confidence': features[2],
                'history_fp_rate': features[3]
            }
        }


# 使用示例
ai_filter = CloudAIFilter()

edge_event = {
    'device_id': 'radar_001',
    'event_type': 'FALL_DETECTED',
    'timestamp': '2024-01-15T08:30:00Z',
    'data': {
        'location': (1.0, 2.0, -2.5),
        'confidence': 0.92,
        'height_ratio': 0.25,
        'point_count': 25
    }
}

device_history = [
    {'is_false_positive': False},
    {'is_false_positive': False},
    {'is_false_positive': True}   # 有一次误报
]

is_true, evaluation = ai_filter.evaluate_event(edge_event, device_history)
print(f"真实事件: {is_true}")
print(f"评估详情: {evaluation}")

4.2 调度中心状态机

python

from enum import Enum
from datetime import datetime
from typing import Optional, List, Dict

class WorkOrderState(Enum):
    """工单状态"""
    CREATED = "created"         # 已创建
    DISPATCHED = "dispatched"   # 已派单
    ACCEPTED = "accepted"       # 已接单
    EN_ROUTE = "en_route"       # 前往中
    ARRIVED = "arrived"         # 已到达
    COMPLETED = "completed"     # 已完成
    ESCALATED = "escalated"     # 已升级
    CANCELLED = "cancelled"     # 已取消


class WorkOrder:
    """服务工单实体"""
    
    def __init__(self, order_id: str, device_id: str, event_type: str, location: str):
        self.order_id = order_id
        self.device_id = device_id
        self.event_type = event_type
        self.location = location
        self.state = WorkOrderState.CREATED
        self.staff_id: Optional[str] = None
        self.timeline: List[Dict] = []
        self._add_timeline("created")
    
    def _add_timeline(self, action: str, detail: Optional[Dict] = None):
        """添加时间线记录"""
        self.timeline.append({
            'timestamp': datetime.now().isoformat(),
            'action': action,
            'detail': detail or {}
        })
    
    def dispatch(self, staff_id: str) -> bool:
        """派单"""
        if self.state != WorkOrderState.CREATED:
            return False
        self.state = WorkOrderState.DISPATCHED
        self.staff_id = staff_id
        self._add_timeline("dispatched", {'staff_id': staff_id})
        return True
    
    def accept(self) -> bool:
        """接单"""
        if self.state != WorkOrderState.DISPATCHED:
            return False
        self.state = WorkOrderState.ACCEPTED
        self._add_timeline("accepted")
        return True
    
    def start_en_route(self) -> bool:
        """开始前往"""
        if self.state != WorkOrderState.ACCEPTED:
            return False
        self.state = WorkOrderState.EN_ROUTE
        self._add_timeline("en_route")
        return True
    
    def arrive(self) -> bool:
        """到达现场"""
        if self.state != WorkOrderState.EN_ROUTE:
            return False
        self.state = WorkOrderState.ARRIVED
        self._add_timeline("arrived")
        return True
    
    def complete(self) -> bool:
        """完成服务"""
        if self.state != WorkOrderState.ARRIVED:
            return False
        self.state = WorkOrderState.COMPLETED
        self._add_timeline("completed")
        return True
    
    def escalate(self, reason: str) -> bool:
        """升级处理（超时或无法处理时）"""
        if self.state in [WorkOrderState.COMPLETED, WorkOrderState.CANCELLED]:
            return False
        self.state = WorkOrderState.ESCALATED
        self._add_timeline("escalated", {'reason': reason})
        return True
    
    def get_response_time_seconds(self) -> Optional[int]:
        """获取从创建到完成的响应时间（秒）"""
        created_time = None
        completed_time = None
        
        for record in self.timeline:
            ts = datetime.fromisoformat(record['timestamp'])
            if record['action'] == 'created':
                created_time = ts
            elif record['action'] == 'completed':
                completed_time = ts
        
        if created_time and completed_time:
            return int((completed_time - created_time).total_seconds())
        return None
    
    def get_full_timeline(self) -> List[Dict]:
        """获取完整时间线"""
        return self.timeline


# 使用示例
order = WorkOrder("WO_001", "radar_001", "FALL_DETECTED", "Room_201")
order.dispatch("staff_001")
order.accept()
order.start_en_route()
order.arrive()
order.complete()

print(f"工单状态: {order.state.value}")
print(f"响应时间: {order.get_response_time_seconds()}秒")
print("\n完整时间线:")
for t in order.get_full_timeline():
    print(f"  {t['timestamp']}: {t['action']}")

---

五、数据驱动的照护运营

5.1 照护排班优化

python

from typing import List, Dict
from datetime import datetime, timedelta

class CareScheduleOptimizer:
    """照护排班优化器"""
    
    def __init__(self):
        self.activity_data: List[Dict] = []
    
    def add_activity_record(self, resident_id: str, activity_type: str, timestamp: datetime):
        """添加活动记录"""
        self.activity_data.append({
            'resident_id': resident_id,
            'activity_type': activity_type,
            'timestamp': timestamp
        })
    
    def get_high_risk_hours(self, resident_id: str) -> List[int]:
        """获取高风险时段（基于历史起夜频率）"""
        night_activities = []
        
        for record in self.activity_data:
            if record['resident_id'] == resident_id and record['activity_type'] == 'night_awakening':
                hour = record['timestamp'].hour
                night_activities.append(hour)
        
        if not night_activities:
            return []
        
        # 统计各时段出现频率
        hour_counts = {}
        for h in night_activities:
            hour_counts[h] = hour_counts.get(h, 0) + 1
        
        # 返回频率最高的3个时段
        sorted_hours = sorted(hour_counts.items(), key=lambda x: x[1], reverse=True)
        return [h for h, _ in sorted_hours[:3]]
    
    def generate_optimized_schedule(self, residents: List[str]) -> Dict[str, List[int]]:
        """生成优化后的巡检排班表"""
        schedule = {}
        
        for resident in residents:
            high_risk_hours = self.get_high_risk_hours(resident)
            schedule[resident] = high_risk_hours
        
        return schedule


# 使用示例
optimizer = CareScheduleOptimizer()

# 模拟起夜记录
for i in range(5):
    optimizer.add_activity_record(
        "resident_001", 
        "night_awakening",
        datetime(2024, 1, 15, 2, 30) + timedelta(hours=i)
    )

high_risk = optimizer.get_high_risk_hours("resident_001")
print(f"高风险时段: {high_risk}点")

5.2 健康数据监测与预警

python

@dataclass
class VitalSign:
    """生命体征数据"""
    resident_id: str
    heart_rate: int      # 心率(次/分)
    respiratory_rate: int # 呼吸频率(次/分)
    systolic_bp: int     # 收缩压(mmHg)
    diastolic_bp: int    # 舒张压(mmHg)
    timestamp: datetime

class HealthMonitor:
    """健康监测器"""
    
    def __init__(self):
        self.vital_signs: List[VitalSign] = []
        self.thresholds = {
            'heart_rate': {'min': 60, 'max': 100},
            'respiratory_rate': {'min': 12, 'max': 20},
            'systolic_bp': {'min': 90, 'max': 140},
            'diastolic_bp': {'min': 60, 'max': 90}
        }
    
    def add_vital_sign(self, vital: VitalSign):
        """添加生命体征记录"""
        self.vital_signs.append(vital)
    
    def check_alerts(self, resident_id: str) -> List[Dict]:
        """检查异常告警"""
        alerts = []
        recent = [v for v in self.vital_signs 
                  if v.resident_id == resident_id 
                  and v.timestamp > datetime.now() - timedelta(days=1)]
        
        if not recent:
            return alerts
        
        latest = recent[-1]
        
        # 检查各项指标
        if latest.heart_rate < self.thresholds['heart_rate']['min']:
            alerts.append({'type': 'BRADYCARDIA', 'value': latest.heart_rate})
        elif latest.heart_rate > self.thresholds['heart_rate']['max']:
            alerts.append({'type': 'TACHYCARDIA', 'value': latest.heart_rate})
        
        if latest.respiratory_rate < self.thresholds['respiratory_rate']['min']:
            alerts.append({'type': 'BRADYPNEA', 'value': latest.respiratory_rate})
        elif latest.respiratory_rate > self.thresholds['respiratory_rate']['max']:
            alerts.append({'type': 'TACHYPNEA', 'value': latest.respiratory_rate})
        
        return alerts


# 使用示例
monitor = HealthMonitor()

vital = VitalSign(
    resident_id="resident_001",
    heart_rate=110,
    respiratory_rate=18,
    systolic_bp=120,
    diastolic_bp=80,
    timestamp=datetime.now()
)
monitor.add_vital_sign(vital)

alerts = monitor.check_alerts("resident_001")
print(f"健康告警: {alerts}")

---

六、技术总结

6.1 关键设计要点

层级	关键设计	技术实现
感知层	边缘预处理，仅上报结构化事件	点云滤波、姿态解算
平台层	云端二次研判，过滤误报	AI评分模型、事件历史
调度层	状态机驱动的工单流转	完整时间线追踪
运营层	数据驱动的排班优化	历史行为分析

6.2 核心指标

指标	目标值	测量方法
跌倒检测准确率	≥95%	实测/标注对比
误报过滤率	≥80%	对比边缘与云端事件
响应时间（派单到完成）	≤10分钟	工单时间线计算
闭环率	≥99%	完成工单/总工单

6.3 架构要点总结

• 边缘侧做第一轮处理，减少云端负载和网络依赖

• 云端AI做二次研判，降低误报率

• 完整的事件状态机和工单时间线，确保可追溯

• 线下服务资源与线上调度深度耦合

注：本文所述技术架构和代码示例为演示用途，具体实现需结合实际硬件选型、网络条件和业务规模进行调整。文中涉及的指标和阈值可根据实际需求配置。#杭州尚宵#智慧养老服务#居家养老解决方案