LLM による意味理解:ハードウェア統合ガイド
概要
LLM モジュールは完全なハードウェア統合フレームワークを提供し、Function Calling インターフェースを通じてさまざまな物理デバイスの制御をサポートします。システムはモジュール設計を採用し、シリアルポートや GPIO などの通信方式をサポートするとともに、安全なハードウェア制御の抽象化レイヤーを提供します。
対応ハードウェアデバイス
ハードウェア機器一覧
1. 6 自由度ロボットアーム
- 型式:汎用 6DOF ロボットアーム
- 通信:USB シリアル通信
- 機能:把持、配置、精密な位置決め
- 制御精度:± 1mm
2. 2 軸パン・チルト
- 型式:汎用サーボ式パン・チルト
- 通信:シリアルポートによる PWM 制御
- 機能:水平/垂直回転
- 角度範囲:0°~270°
3. オーディオ再生機器
- インターフェース:システムのオーディオ出力
- 形式:MP3、WAV、OGG など
- 機能:再生制御、音量調整
4. カメラ機器
- インターフェース:USB/CSI カメラ
- 機能:画像取得、視覚認識
- 解像度:複数の解像度に対応
ロボットアーム制御システム
システムアーキテクチャ
+-------------------+
| LLM指令处理 |
+-------------------+
|
+-------------------+
| 机械臂函数接口 |
+-------------------+
|
+-------------------+
| RobotArm类 |
+-------------------+
|
+----+----+
| |
|串口通信 | 运动规划算法
|协议 |
+----+----+
|
+-------------------+
| 物理机械臂 |
+-------------------+
コアクラスの定義
class RobotArm:
"""6自由度机械臂控制类"""
def __init__(self, port: str = "/dev/ttyS1", baudrate: int = 115200):
"""
初始化机械臂控制器
Args:
port: 串口设备路径
baudrate: 波特率
"""
self.port = port
self.baudrate = baudrate
self.serial_conn = None
self.current_position = [0, 0, 0] # X, Y, Z
self.current_angles = [0, 0, 0, 0, 0, 0] # 6个关节角度
self.is_grabbing = False
def connect(self) -> bool:
"""连接机械臂"""
try:
self.serial_conn = serial.Serial(self.port, self.baudrate, timeout=2.0)
time.sleep(2) # 等待初始化
return self.serial_conn.is_open
except Exception as e:
logger.error(f"机械臂连接失败: {e}")
return False
def move_to_position(self, x: float, y: float, z: float, speed: float = 50.0) -> bool:
"""移动到指定位置"""
# 安全检查
if not self.is_position_safe(x, y, z):
logger.warning(f"位置超出安全范围: ({x}, {y}, {z})")
return False
# 逆运动学计算
joint_angles = self.inverse_kinematics(x, y, z)
if joint_angles is None:
logger.error("逆运动学求解失败")
return False
# 发送运动指令
success = self.move_joints(joint_angles, speed)
if success:
self.current_position = [x, y, z]
self.current_angles = joint_angles
return success
def grab_object(self, x: float, y: float, z: float) -> bool:
"""在指定位置抓取物体"""
try:
# 移动到目标上方
if not self.move_to_position(x, y, z + 50):
return False
# 下降到目标位置
if not self.move_to_position(x, y, z):
return False
# 闭合夹爪
if not self.close_gripper():
return False
# 提升物体
if not self.move_to_position(x, y, z + 50):
return False
self.is_grabbing = True
logger.info(f"成功抓取物体于位置: ({x}, {y}, {z})")
return True
except Exception as e:
logger.error(f"抓取物体失败: {e}")
return False
運動学アルゴリズム
順運動学
def forward_kinematics(self, joint_angles: List[float]) -> Tuple[float, float, float]:
"""
正运动学计算:根据关节角度计算末端位置
Args:
joint_angles: 6个关节角度 (度)
Returns:
末端执行器的 (x, y, z) 坐标
"""
# 机械臂DH参数 (Denavit-Hartenberg parameters)
dh_params = [
[0, 90, 85, 0], # 关节1
[105, 0, 0, -90], # 关节2
[95, 0, 0, 0], # 关节3
[0, 90, 82, 0], # 关节4
[0, -90, 0, 0], # 关节5
[0, 0, 45, 0] # 关节6
]
# 转换为弧度
angles_rad = [math.radians(angle) for angle in joint_angles]
# 计算变换矩阵
T = np.eye(4)
for i, (a, alpha, d, theta_offset) in enumerate(dh_params):
theta = angles_rad[i] + math.radians(theta_offset)
# DH变换矩阵
Ti = np.array([
[math.cos(theta), -math.sin(theta)*math.cos(math.radians(alpha)),
math.sin(theta)*math.sin(math.radians(alpha)), a*math.cos(theta)],
[math.sin(theta), math.cos(theta)*math.cos(math.radians(alpha)),
-math.cos(theta)*math.sin(math.radians(alpha)), a*math.sin(theta)],
[0, math.sin(math.radians(alpha)),
math.cos(math.radians(alpha)), d],
[0, 0, 0, 1]
])
T = T @ Ti
return T[0, 3], T[1, 3], T[2, 3]
逆運動学
def inverse_kinematics(self, x: float, y: float, z: float) -> Optional[List[float]]:
"""
逆运动学求解:根据目标位置计算关节角度
"""
try:
from scipy.optimize import fsolve
def equations(angles):
fx, fy, fz = self.forward_kinematics(angles)
return [fx - x, fy - y, fz - z]
initial_guess = self.current_angles.copy()
solution = fsolve(equations, initial_guess)
# 验证解的有效性
error_x, error_y, error_z = equations(solution)
error_magnitude = math.sqrt(error_x**2 + error_y**2 + error_z**2)
if error_magnitude < 1.0 and self.are_angles_valid(solution):
return solution.tolist()
return None
except Exception as e:
logger.error(f"逆运动学计算错误: {e}")
return None
安全制御システム
class SafetyController:
"""机械臂安全控制器"""
def __init__(self, arm: RobotArm):
self.arm = arm
self.workspace_limits = {
'x': [-200, 200], # mm
'y': [-200, 200], # mm
'z': [0, 300] # mm
}
self.joint_limits = {
'joint1': [-180, 180], # 度
'joint2': [-90, 90],
'joint3': [-90, 90],
'joint4': [-180, 180],
'joint5': [-90, 90],
'joint6': [-180, 180]
}
self.max_speed = 100 # mm/s
self.emergency_stop = False
def is_position_safe(self, x: float, y: float, z: float) -> bool:
"""检查位置是否在安全范围内"""
return (self.workspace_limits['x'][0] <= x <= self.workspace_limits['x'][1] and
self.workspace_limits['y'][0] <= y <= self.workspace_limits['y'][1] and
self.workspace_limits['z'][0] <= z <= self.workspace_limits['z'][1])
def emergency_stop_all(self):
"""紧急停止所有运动"""
self.emergency_stop = True
try:
if self.arm.serial_conn and self.arm.serial_conn.is_open:
self.arm.serial_conn.write(b"EMERGENCY_STOP\n")
logger.critical("机械臂紧急停止已激活")
except Exception as e:
logger.error(f"紧急停止指令发送失败: {e}")
パン・チルト制御システム
パン・チルトコントローラ
class GimbalController:
"""双轴云台控制器"""
def __init__(self, port: str, baudrate: int, pan_pin: int, tilt_pin: int):
"""
初始化云台控制器
Args:
port: 串口设备路径
baudrate: 波特率
pan_pin: 水平舵机控制引脚
tilt_pin: 垂直舵机控制引脚
"""
self.port = port
self.baudrate = baudrate
self.pan_pin = pan_pin
self.tilt_pin = tilt_pin
self.serial_conn = None
self.current_pan = 135.0 # 中心位置
self.current_tilt = 135.0 # 中心位置
self.angle_range = [0, 270] # 角度范围
def connect(self) -> bool:
"""连接云台控制器"""
try:
self.serial_conn = serial.Serial(self.port, self.baudrate, timeout=1.0)
time.sleep(1)
return self.serial_conn.is_open
except Exception as e:
logger.error(f"云台连接失败: {e}")
return False
def set_position(self, pan: float, tilt: float) -> bool:
"""
设置云台位置
Args:
pan: 水平角度 (0 到 270 度)
tilt: 垂直角度 (0 到 270 度)
Returns:
设置是否成功
"""
# 角度限制
pan = max(self.angle_range[0], min(self.angle_range[1], pan))
tilt = max(self.angle_range[0], min(self.angle_range[1], tilt))
try:
# 发送控制指令
command = f"G1 X{pan} Y{tilt}\n"
self.serial_conn.write(command.encode())
# 等待响应
response = self.serial_conn.readline().decode().strip()
if response == "OK":
self.current_pan = pan
self.current_tilt = tilt
logger.info(f"云台位置设置: Pan={pan}°, Tilt={tilt}°")
return True
else:
logger.error(f"云台控制失败: {response}")
return False
except Exception as e:
logger.error(f"云台控制错误: {e}")
return False
def move_relative(self, pan_delta: float, tilt_delta: float) -> bool:
"""相对移动"""
new_pan = self.current_pan + pan_delta
new_tilt = self.current_tilt + tilt_delta
return self.set_position(new_pan, new_tilt)
def reset_position(self) -> bool:
"""重置到中心位置"""
return self.set_position(135.0, 135.0)
パン・チルトのキャリブレーションシステム
class GimbalCalibration:
"""云台标定系统"""
def __init__(self, gimbal: GimbalController):
self.gimbal = gimbal
self.calibration_data = {}
def run_calibration(self) -> bool:
"""运行标定程序"""
logger.info("开始云台标定...")
# 标定点列表
calibration_points = [
(0, 0), (0, 135), (0, 270),
(135, 0), (135, 135), (135, 270),
(270, 0), (270, 135), (270, 270)
]
for pan, tilt in calibration_points:
logger.info(f"标定点: ({pan}, {tilt})")
if not self.gimbal.set_position(pan, tilt):
logger.error(f"无法移动到标定点 ({pan}, {tilt})")
return False
# 等待稳定
time.sleep(2)
# 记录实际位置
actual_pan, actual_tilt = self.measure_actual_position()
self.calibration_data[(pan, tilt)] = (actual_pan, actual_tilt)
self.calculate_calibration_parameters()
logger.info("云台标定完成")
return True
def measure_actual_position(self) -> Tuple[float, float]:
"""测量实际位置"""
return self.gimbal.current_pan, self.gimbal.current_tilt
def calculate_calibration_parameters(self):
"""计算标定参数"""
pan_errors = []
tilt_errors = []
for (cmd_pan, cmd_tilt), (actual_pan, actual_tilt) in self.calibration_data.items():
pan_errors.append(actual_pan - cmd_pan)
tilt_errors.append(actual_tilt - cmd_tilt)
self.pan_offset = sum(pan_errors) / len(pan_errors)
self.tilt_offset = sum(tilt_errors) / len(tilt_errors)
logger.info(f"标定偏移: Pan={self.pan_offset:.2f}°, Tilt={self.tilt_offset:.2f}°")
オーディオシステムの統合
オーディオプレーヤー
class AudioPlayer:
"""音频播放器"""
def __init__(self, music_directory: str = "./music"):
"""
初始化音频播放器
Args:
music_directory: 音乐文件目录
"""
self.music_directory = Path(music_directory)
self.current_music = None
self.is_playing = False
self.volume = 0.7
self.supported_formats = ['.mp3', '.wav', '.ogg', '.m4a']
# 初始化pygame音频系统
pygame.mixer.pre_init(frequency=22050, size=-16, channels=2, buffer=512)
pygame.mixer.init()
def list_music(self) -> List[str]:
"""获取音乐文件列表"""
if not self.music_directory.exists():
self.music_directory.mkdir(parents=True)
return []
music_files = []
for ext in self.supported_formats:
music_files.extend(self.music_directory.glob(f"*{ext}"))
return [f.name for f in music_files]
def play_music(self, music_name: str) -> bool:
"""播放指定音乐"""
music_file = self.find_music_file(music_name)
if not music_file:
logger.warning(f"未找到音乐文件: {music_name}")
return False
try:
self.stop_music()
pygame.mixer.music.load(str(music_file))
pygame.mixer.music.set_volume(self.volume)
pygame.mixer.music.play()
self.current_music = music_name
self.is_playing = True
logger.info(f"开始播放音乐: {music_name}")
return True
except Exception as e:
logger.error(f"音乐播放失败: {e}")
return False
def stop_music(self):
"""停止音乐播放"""
try:
pygame.mixer.music.stop()
self.is_playing = False
self.current_music = None
logger.info("音乐播放已停止")
except Exception as e:
logger.error(f"停止音乐失败: {e}")
def set_volume(self, volume: float):
"""设置音量"""
self.volume = max(0.0, min(1.0, volume))
pygame.mixer.music.set_volume(self.volume)
logger.info(f"音量设置为: {self.volume * 100:.0f}%")
def get_status(self) -> dict:
"""获取播放状态"""
return {
"is_playing": self.is_playing,
"current_music": self.current_music,
"volume": self.volume,
"music_count": len(self.list_music())
}
ビジョンシステムの統合
カメラ制御
class CameraController:
"""摄像头控制器"""
def __init__(self, device_id: int = 0, resolution: tuple = (640, 480)):
"""
初始化摄像头
Args:
device_id: 摄像头设备ID
resolution: 分辨率 (宽, 高)
"""
self.device_id = device_id
self.resolution = resolution
self.cap = None
self.is_opened = False
def open_camera(self) -> bool:
"""打开摄像头"""
try:
self.cap = cv2.VideoCapture(self.device_id)
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.resolution[0])
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.resolution[1])
self.is_opened = self.cap.isOpened()
if self.is_opened:
logger.info(f"摄像头已打开: 设备{self.device_id}, 分辨率{self.resolution}")
else:
logger.error(f"无法打开摄像头设备: {self.device_id}")
return self.is_opened
except Exception as e:
logger.error(f"摄像头打开失败: {e}")
return False
def capture_image(self, save_path: str = None) -> Optional[np.ndarray]:
"""捕获图像"""
if not self.is_opened:
logger.warning("摄像头未打开")
return None
try:
ret, frame = self.cap.read()
if ret:
if save_path:
cv2.imwrite(save_path, frame)
logger.info(f"图像已保存到: {save_path}")
return frame
else:
logger.error("图像捕获失败")
return None
except Exception as e:
logger.error(f"图像捕获错误: {e}")
return None
def close_camera(self):
"""关闭摄像头"""
if self.cap:
self.cap.release()
self.is_opened = False
logger.info("摄像头已关闭")
視覚認識の統合
class VisionSystem:
"""视觉识别系统"""
def __init__(self, camera: CameraController, vlm_client):
self.camera = camera
self.vlm_client = vlm_client
self.temp_image_dir = Path("./temp_images")
self.temp_image_dir.mkdir(exist_ok=True)
def analyze_current_view(self, query: str = "请描述看到的内容") -> str:
"""分析当前摄像头画面"""
timestamp = int(time.time())
image_path = self.temp_image_dir / f"capture_{timestamp}.jpg"
image = self.camera.capture_image(str(image_path))
if image is None:
return "无法获取摄像头图像"
try:
result = self.vlm_client.analyze_image(str(image_path), query)
image_path.unlink(missing_ok=True) # 清理临时文件
return result
except Exception as e:
logger.error(f"视觉分析失败: {e}")
return f"视觉分析错误: {str(e)}"
def detect_objects_in_view(self) -> List[dict]:
"""检测当前视野中的物体"""
timestamp = int(time.time())
image_path = self.temp_image_dir / f"detect_{timestamp}.jpg"
image = self.camera.capture_image(str(image_path))
if image is None:
return []
try:
objects = self.vlm_client.detect_objects(str(image_path))
image_path.unlink(missing_ok=True) # 清理临时文件
return objects
except Exception as e:
logger.error(f"物体检测失败: {e}")
return []
⚙️ ハードウェア設定管理
デバイス設定ファイル
# 硬件设备配置
hardware:
# 机械臂配置
robot_arm:
enabled: true
port: "/dev/ttyS1"
baudrate: 115200
timeout: 5.0
# 工作空间限制 (mm)
workspace:
x_range: [-200, 200]
y_range: [-200, 200]
z_range: [0, 300]
# 关节角度限制 (度)
joint_limits:
joint1: [-180, 180]
joint2: [-90, 90]
joint3: [-90, 90]
joint4: [-180, 180]
joint5: [-90, 90]
joint6: [-180, 180]
# 运动参数
max_speed: 100 # mm/s
acceleration: 50 # mm/s²
# 云台配置
gimbal:
enabled: true
port: "/dev/ttyS1"
baudrate: 115200
timeout: 3.0
# 舵机引脚
pan_pin: 3
tilt_pin: 4
# 角度限制
angle_range: [0, 270]
init_pan_angle: 135.0
init_tilt_angle: 135.0
# 音频配置
audio:
enabled: true
music_directory: "./music"
default_volume: 0.7
supported_formats: [".mp3", ".wav", ".ogg", ".m4a"]
# 摄像头配置
camera:
enabled: true
device_id: 0
resolution: [640, 480]
fps: 30
# 安全配置
safety:
emergency_stop_enabled: true
collision_detection: true
workspace_monitoring: true
# 监控参数
position_check_interval: 0.1 # 秒
force_threshold: 50 # N
temperature_threshold: 70 # °C
デバイスマネージャー
class HardwareManager:
"""硬件设备管理器"""
def __init__(self, config: dict):
self.config = config
self.devices = {}
self.safety_monitor = SafetyMonitor()
def initialize_devices(self) -> bool:
"""初始化所有硬件设备"""
success = True
# 初始化机械臂
if self.config['hardware']['robot_arm']['enabled']:
success &= self._init_robot_arm()
# 初始化云台
if self.config['hardware']['gimbal']['enabled']:
success &= self._init_gimbal()
# 初始化音频系统
if self.config['hardware']['audio']['enabled']:
success &= self._init_audio()
# 初始化摄像头
if self.config['hardware']['camera']['enabled']:
success &= self._init_camera()
# 启动安全监控
if success:
self.safety_monitor.start()
return success
def get_device(self, device_name: str):
"""获取设备实例"""
return self.devices.get(device_name)
def shutdown_all(self):
"""关闭所有设备"""
for device_name, device in self.devices.items():
try:
if hasattr(device, 'disconnect'):
device.disconnect()
elif hasattr(device, 'close'):
device.close()
logger.info(f"设备已关闭: {device_name}")
except Exception as e:
logger.error(f"设备关闭失败 {device_name}: {e}")
self.safety_monitor.stop()
トラブルシューティングガイド
よくあるハードウェアの問題
- シリアルデバイスにアクセスできない
# 检查设备文件
ls -l /dev/ttyS*
# 添加用户到dialout组
sudo usermod -a -G dialout $USER
# 重新登录或重启
- ロボットアームの通信障害
- シリアル接続を確認します
- ボーレートが正しく設定されているか(115200)を確認します
- 機器の電源を確認します
- 通信プロトコルが一致しているかを確認します
- パン・チルト制御の異常
- サーボの電源電圧を確認します
- ピンが正しく接続されているかを確認します
- 角度範囲の設定を確認します
- 制御ボードの状態を確認します
- オーディオ再生の問題
# 检查音频系统
aplay -l
# 安装pygame
pip install pygame==2.5.2
# 检查音频文件格式
file ./music/*.mp3
デバッグツール
class HardwareDebugger:
"""硬件调试工具"""
def __init__(self, hardware_manager: HardwareManager):
self.hardware_manager = hardware_manager
def run_diagnostic(self) -> dict:
"""运行硬件诊断"""
results = {}
# 机械臂诊断
if 'robot_arm' in self.hardware_manager.devices:
results['robot_arm'] = self._test_robot_arm()
# 云台诊断
if 'gimbal' in self.hardware_manager.devices:
results['gimbal'] = self._test_gimbal()
return results
def _test_robot_arm(self) -> dict:
"""测试机械臂功能"""
arm = self.hardware_manager.get_device('robot_arm')
tests = {}
# 连接测试
tests['connection'] = arm.serial_conn.is_open if arm.serial_conn else False
# 运动测试
try:
original_pos = arm.current_position.copy()
tests['movement'] = arm.move_to_position(
original_pos[0] + 10,
original_pos[1],
original_pos[2]
)
arm.move_to_position(*original_pos) # 回到原位置
except Exception:
tests['movement'] = False
return tests
このハードウェア統合システムは、LLM モジュールに完全な物理デバイス制御機能を提供します。簡単なオーディオ再生から複雑なロボットアーム制御まで、自然言語の指示によって正確に制御できます。
ドキュメント