003_具身人工智能的核心组件与安全需求：从感知决策到执行的全方位安全保障

# 传感器异常检测示例代码
class SensorAnomalyDetector:
    def __init__(self, sensor_types, thresholds):
        self.sensor_types = sensor_types
        self.thresholds = thresholds
        self.history_data = {sensor: [] for sensor in sensor_types}
    
    def detect_anomaly(self, sensor_data):
        anomalies = []
        
        for sensor, data in sensor_data.items():
            if sensor in self.thresholds:
                # 检查数据范围异常
                if not (self.thresholds[sensor]['min'] <= data <= self.thresholds[sensor]['max']):
                    anomalies.append({
                        'sensor': sensor,
                        'type': 'range_anomaly',
                        'value': data,
                        'threshold': f"[{self.thresholds[sensor]['min']}, {self.thresholds[sensor]['max']}]"
                    })
                
                # 检查数据变化率异常
                if self.history_data[sensor]:
                    rate_of_change = abs(data - self.history_data[sensor][-1])
                    if rate_of_change > self.thresholds[sensor]['max_rate']:
                        anomalies.append({
                            'sensor': sensor,
                            'type': 'rate_anomaly',
                            'value': rate_of_change,
                            'threshold': self.thresholds[sensor]['max_rate']
                        })
                
                # 更新历史数据
                self.history_data[sensor].append(data)
                if len(self.history_data[sensor]) > 100:  # 保持最近100个数据点
                    self.history_data[sensor] = self.history_data[sensor][-100:]
        
        return anomalies

# 使用示例
anomaly_detector = SensorAnomalyDetector(
    ['lidar', 'camera', 'imu'],
    {
        'lidar': {'min': 0, 'max': 100, 'max_rate': 5},
        'camera': {'min': 0, 'max': 255, 'max_rate': 100},
        'imu': {'min': -10, 'max': 10, 'max_rate': 2}
    }
)

# 安全强化学习示例框架
class SafeReinforcementLearning:
    def __init__(self, env, safety_constraints):
        self.env = env
        self.safety_constraints = safety_constraints
        self.policy_network = self._build_policy_network()
        self.safety_critic = self._build_safety_critic()
    
    def _build_policy_network(self):
        # 构建策略网络
        pass
    
    def _build_safety_critic(self):
        # 构建安全评估网络
        pass
    
    def check_safety_constraints(self, state, action):
        # 检查安全约束
        for constraint in self.safety_constraints:
            if not constraint(state, action):
                return False, constraint.__name__
        return True, ""
    
    def get_safe_action(self, state):
        # 获取符合安全约束的动作
        raw_action = self.policy_network.predict(state)
        is_safe, violated_constraint = self.check_safety_constraints(state, raw_action)
        
        if not is_safe:
            # 采取安全回退动作
            safe_action = self._generate_safe_fallback(state, violated_constraint)
            return safe_action, True  # True表示触发了安全回退
        
        return raw_action, False
    
    def _generate_safe_fallback(self, state, violated_constraint):
        # 根据违反的约束生成安全回退动作
        # 例如：降低速度、停止运动、返回安全位置等
        pass

# 执行模块安全监控示例
class ExecutionSafetyMonitor:
    def __init__(self, actuator_configs):
        self.actuator_configs = actuator_configs
        self.safety_limits = {}
        for actuator, config in actuator_configs.items():
            self.safety_limits[actuator] = {
                'position_min': config['position_min'],
                'position_max': config['position_max'],
                'velocity_max': config['velocity_max'],
                'current_max': config['current_max'],
                'temperature_max': config['temperature_max']
            }
    
    def monitor_execution(self, actuator_states):
        safety_alerts = []
        emergency_stop_required = False
        
        for actuator, state in actuator_states.items():
            if actuator in self.safety_limits:
                limits = self.safety_limits[actuator]
                
                # 检查位置限制
                if state['position'] < limits['position_min'] or state['position'] > limits['position_max']:
                    alerts = self._generate_position_alert(actuator, state, limits)
                    safety_alerts.extend(alerts)
                    emergency_stop_required = True
                
                # 检查速度限制
                if abs(state['velocity']) > limits['velocity_max']:
                    alerts = self._generate_velocity_alert(actuator, state, limits)
                    safety_alerts.extend(alerts)
                    emergency_stop_required = True
                
                # 检查电流限制
                if state['current'] > limits['current_max']:
                    alerts = self._generate_current_alert(actuator, state, limits)
                    safety_alerts.extend(alerts)
                    if state['current'] > 1.5 * limits['current_max']:  # 严重过流立即停止
                        emergency_stop_required = True
                
                # 检查温度限制
                if state['temperature'] > limits['temperature_max']:
                    alerts = self._generate_temperature_alert(actuator, state, limits)
                    safety_alerts.extend(alerts)
        
        return {
            'alerts': safety_alerts,
            'emergency_stop_required': emergency_stop_required,
            'timestamp': self._get_current_timestamp()
        }
    
    def _generate_position_alert(self, actuator, state, limits):
        # 生成位置越界警报
        pass
    
    # 其他警报生成方法...

# 安全通信协议示例
class SecureCommunication:
    def __init__(self, cert_path, key_path, ca_cert_path):
        self.cert_path = cert_path
        self.key_path = key_path
        self.ca_cert_path = ca_cert_path
        self.secure_channel = self._initialize_secure_channel()
    
    def _initialize_secure_channel(self):
        # 初始化安全通信通道
        # 配置TLS上下文，加载证书
        pass
    
    def send_secure_data(self, data, destination):
        # 加密并发送数据
        # 1. 数据完整性校验
        # 2. 数据加密
        # 3. 发送并验证接收
        pass
    
    def receive_secure_data(self):
        # 接收并解密数据
        # 1. 接收数据
        # 2. 验证数据完整性
        # 3. 解密数据
        pass
    
    def verify_component_identity(self, component_id):
        # 验证通信组件身份
        # 基于证书链验证
        pass

class EmbodiedAISecurityTester:
    def __init__(self, system_under_test):
        self.system = system_under_test
        self.test_results = []
    
    def run_perception_security_tests(self):
        # 运行感知模块安全测试
        tests = [
            self._test_sensor_spoofing,
            self._test_adversarial_attacks,
            self._test_sensor_noise_resistance
        ]
        return self._execute_tests(tests)
    
    def run_decision_security_tests(self):
        # 运行决策模块安全测试
        tests = [
            self._test_model_poisoning_resistance,
            self._test_decision_boundaries,
            self._test_safety_constraints
        ]
        return self._execute_tests(tests)
    
    def run_execution_security_tests(self):
        # 运行执行模块安全测试
        tests = [
            self._test_emergency_stop,
            self._test_physical_constraints,
            self._test_fault_recovery
        ]
        return self._execute_tests(tests)
    
    def _execute_tests(self, tests):
        results = []
        for test in tests:
            try:
                result = test()
                results.append({
                    'test': test.__name__,
                    'passed': result['passed'],
                    'details': result['details']
                })
            except Exception as e:
                results.append({
                    'test': test.__name__,
                    'passed': False,
                    'error': str(e)
                })
        return results