001_具身人工智能概述与安全引言：从理论基础到前沿应用的深度解析

# ROS 2安全插件示例：安全节点配置

from rclpy.node import Node
from rclpy.security import SecurityContext

class SecureRobotNode(Node):
    def __init__(self):
        # 创建带有安全上下文的节点
        security_context = SecurityContext()
        security_context.load_security_files(
            certificate_path='/path/to/cert.pem',
            private_key_path='/path/to/key.pem',
            ca_cert_path='/path/to/ca.pem'
        )
        
        super().__init__('secure_robot_node', security_context=security_context)
        
        # 设置安全订阅器，验证发布者身份
        self.subscription = self.create_subscription(
            String,  # 消息类型
            'sensor_data',  # 话题名
            self.listener_callback,
            10,  # QoS
            callback_group=None,
            raw=False
        )
        
        # 设置安全发布器，签名消息
        self.publisher = self.create_publisher(
            String,  # 消息类型
            'control_commands',  # 话题名
            10,  # QoS
            callback_group=None
        )
    
    def listener_callback(self, msg):
        # 验证消息完整性
        if self.verify_message_integrity(msg):
            # 处理验证通过的消息
            processed_data = self.process_sensor_data(msg.data)
            # 发布控制命令
            self.publish_command(processed_data)
    
    def verify_message_integrity(self, msg):
        # 实现消息完整性验证逻辑
        # 1. 验证消息签名
        # 2. 检查消息时间戳是否合理
        # 3. 检查数据范围是否在预期内
        return True  # 简化示例
    
    def process_sensor_data(self, data):
        # 实现传感器数据处理逻辑
        return f"processed:{data}"
    
    def publish_command(self, command):
        # 发布签名的控制命令
        msg = String()
        msg.data = command
        # 消息将自动使用节点的安全上下文进行签名
        self.publisher.publish(msg)
        self.get_logger().info(f'发布安全命令: {command}')

# 主函数
import rclpy
from std_msgs.msg import String

def main(args=None):
    rclpy.init(args=args)
    
    # 创建并配置安全环境
    secure_node = SecureRobotNode()
    
    try:
        # 运行节点
        rclpy.spin(secure_node)
    except KeyboardInterrupt:
        secure_node.get_logger().info('收到退出信号，关闭节点')
    finally:
        # 清理资源
        secure_node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

# 具身AI传感器数据异常检测工具

import numpy as np
import tensorflow as tf
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

class SensorAnomalyDetector:
    def __init__(self, model_type='iforest'):
        """初始化传感器异常检测模型
        
        参数:
            model_type: 模型类型，'iforest'或'autoencoder'
        """
        self.model_type = model_type
        self.scaler = StandardScaler()
        
        if model_type == 'iforest':
            # 隔离森林模型
            self.model = IsolationForest(
                n_estimators=100,
                contamination=0.01,
                random_state=42
            )
        elif model_type == 'autoencoder':
            # 自编码器模型
            self.model = self._build_autoencoder(input_dim=6)
        else:
            raise ValueError(f'不支持的模型类型: {model_type}')
        
        self.is_trained = False
    
    def _build_autoencoder(self, input_dim=6):
        """构建自编码器模型"""
        inputs = tf.keras.layers.Input(shape=(input_dim,))
        
        # 编码器
        encoded = tf.keras.layers.Dense(128, activation='relu')(inputs)
        encoded = tf.keras.layers.Dense(64, activation='relu')(encoded)
        encoded = tf.keras.layers.Dense(32, activation='relu')(encoded)
        
        # 解码器
        decoded = tf.keras.layers.Dense(64, activation='relu')(encoded)
        decoded = tf.keras.layers.Dense(128, activation='relu')(decoded)
        decoded = tf.keras.layers.Dense(input_dim, activation='linear')(decoded)
        
        # 构建自编码器模型
        autoencoder = tf.keras.models.Model(inputs, decoded)
        autoencoder.compile(optimizer='adam', loss='mse')
        
        return autoencoder
    
    def fit(self, X_train):
        """训练异常检测模型
        
        参数:
            X_train: 训练数据，形状为 (n_samples, n_features)
        """
        # 数据标准化
        X_scaled = self.scaler.fit_transform(X_train)
        
        if self.model_type == 'iforest':
            # 训练隔离森林
            self.model.fit(X_scaled)
        elif self.model_type == 'autoencoder':
            # 训练自编码器
            self.model.fit(
                X_scaled, X_scaled,
                epochs=100,
                batch_size=32,
                shuffle=True,
                validation_split=0.2,
                verbose=0
            )
            # 计算重建误差阈值
            X_pred = self.model.predict(X_scaled)
            mse = np.mean(np.power(X_scaled - X_pred, 2), axis=1)
            self.threshold = np.percentile(mse, 99)  # 99%分位数作为阈值
        
        self.is_trained = True
        print(f"模型训练完成，类型: {self.model_type}")
    
    def detect_anomalies(self, X_test, return_scores=False):
        """检测异常数据
        
        参数:
            X_test: 测试数据，形状为 (n_samples, n_features)
            return_scores: 是否返回异常分数
            
        返回:
            predictions: 预测结果，1表示正常，-1表示异常
            scores: 异常分数 (如果return_scores为True)
        """
        if not self.is_trained:
            raise ValueError("模型尚未训练，请先调用fit方法")
        
        # 数据标准化
        X_scaled = self.scaler.transform(X_test)
        
        if self.model_type == 'iforest':
            # 使用隔离森林预测
            predictions = self.model.predict(X_scaled)
            scores = self.model.score_samples(X_scaled)  # 异常分数
        elif self.model_type == 'autoencoder':
            # 使用自编码器预测
            X_pred = self.model.predict(X_scaled)
            mse = np.mean(np.power(X_scaled - X_pred, 2), axis=1)  # 重建误差
            predictions = np.where(mse > self.threshold, -1, 1)  # -1表示异常
            scores = mse
        
        if return_scores:
            return predictions, scores
        else:
            return predictions
    
    def get_anomaly_metrics(self, X_test, y_true):
        """计算异常检测的性能指标
        
        参数:
            X_test: 测试数据
            y_true: 真实标签，1表示正常，-1表示异常
            
        返回:
            metrics: 性能指标字典
        """
        from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
        
        predictions = self.detect_anomalies(X_test)
        
        metrics = {
            'accuracy': accuracy_score(y_true, predictions),
            'precision': precision_score(y_true, predictions, pos_label=-1),
            'recall': recall_score(y_true, predictions, pos_label=-1),
            'f1_score': f1_score(y_true, predictions, pos_label=-1)
        }
        
        return metrics

# 使用示例
if __name__ == "__main__":
    # 生成示例数据
    np.random.seed(42)
    # 正常数据 - 多维高斯分布
    n_samples = 1000
    X_normal = np.random.randn(n_samples, 6)
    # 异常数据 - 离群点
    n_anomalies = 50
    X_anomalies = 10 + np.random.randn(n_anomalies, 6)  # 远离原点的点
    
    # 合并数据集
    X_train = X_normal[:800]
    X_test = np.vstack([X_normal[800:], X_anomalies])
    y_test = np.concatenate([np.ones(200), -np.ones(n_anomalies)])
    
    # 创建并训练检测器
    detector = SensorAnomalyDetector(model_type='autoencoder')
    detector.fit(X_train)
    
    # 检测异常
    predictions, scores = detector.detect_anomalies(X_test, return_scores=True)
    
    # 计算性能指标
    metrics = detector.get_anomaly_metrics(X_test, y_test)
    print("异常检测性能指标:")
    for key, value in metrics.items():
        print(f"{key}: {value:.4f}")
    
    # 显示检测结果统计
    n_true_anomalies = (y_test == -1).sum()
    n_detected_anomalies = (predictions == -1).sum()
    n_correctly_detected = ((predictions == -1) & (y_test == -1)).sum()
    
    print(f"\n测试集中异常样本数量: {n_true_anomalies}")
    print(f"检测到的异常样本数量: {n_detected_anomalies}")
    print(f"正确检测的异常样本数量: {n_correctly_detected}")
    print(f"异常检测率: {n_correctly_detected / n_true_anomalies:.4f}")