Motion Planning Application Service

controller.application.services.motion_planning_application_service

运动规划应用服务

MotionPlanningApplicationService

MotionPlanningApplicationService(domain_service: MotionPlanningDomainService, repository: MotionPlanRepository, motion_constructor: MotionConstructor, command_hub: CommandHubService, message_response: MessageResponseService, trajectory_repository: TrajectoryRepository)

Bases: QObject

运动规划应用服务。

职责： 1. 协调Domain和Infrastructure 2. 自动保存 3. 发送UI更新信号 4. 执行运动规划（单点和整体方案） 5. 处理"获取位置"功能

属性：

名称	类型	描述
plan_list_changed	pyqtSignal	方案列表变化信号。
current_plan_changed	pyqtSignal	当前方案切换信号。
point_list_changed	pyqtSignal	节点列表变化信号。
current_position_received	pyqtSignal	当前位置数据信号（弧度值）。
trajectory_preview_signal	pyqtSignal	轨迹预览数据信号（轨迹数据，上下文）。

初始化运动规划应用服务。

参数：

名称	类型	描述	默认
domain_service	MotionPlanningDomainService	运动规划领域服务。	必需
repository	MotionPlanRepository	运动方案仓储。	必需
motion_constructor	MotionConstructor	运动构造器。	必需
command_hub	CommandHubService	命令中心服务。	必需
message_response	MessageResponseService	消息响应服务。	必需
trajectory_repository	TrajectoryRepository	轨迹仓储。	必需

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def __init__(
    self,
    domain_service: MotionPlanningDomainService,
    repository: MotionPlanRepository,
    motion_constructor: MotionConstructor,
    command_hub: CommandHubService,
    message_response: MessageResponseService,
    trajectory_repository: TrajectoryRepository
):
    """初始化运动规划应用服务。

    Args:
        domain_service: 运动规划领域服务。
        repository: 运动方案仓储。
        motion_constructor: 运动构造器。
        command_hub: 命令中心服务。
        message_response: 消息响应服务。
        trajectory_repository: 轨迹仓储。
    """
    super().__init__()
    self.domain_service = domain_service
    self.repository = repository
    self.motion_constructor = motion_constructor
    self.command_hub = command_hub
    self.message_response = message_response
    self.trajectory_repository = trajectory_repository

    # 连接MessageResponse的位置数据信号
    self.message_response.get_current_position_signal.connect(
        self.current_position_received.emit
    )

    # 连接MessageResponse的轨迹预览信号
    self.message_response.trajectory_preview_signal.connect(
        self.trajectory_preview_signal.emit
    )

    self._load_data()

create_plan

create_plan(name: str)

创建方案。

参数：

名称	类型	描述	默认
name	str	方案名称。	必需

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def create_plan(self, name: str):
    """创建方案。

    Args:
        name (str): 方案名称。
    """
    new_index = self.domain_service.create_plan(name)
    self._save_data()
    self.plan_list_changed.emit()
    self.current_plan_changed.emit(new_index)
    self.point_list_changed.emit()

delete_plan

delete_plan(index: int) -> bool

删除方案。

参数：

名称	类型	描述	默认
index	int	方案索引。	必需

返回：

名称	类型	描述
bool	bool	True=删除成功, False=删除失败（违反业务规则）。

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def delete_plan(self, index: int) -> bool:
    """删除方案。

    Args:
        index (int): 方案索引。

    Returns:
        bool: True=删除成功, False=删除失败（违反业务规则）。
    """
    if self.domain_service.delete_plan(index):
        self._save_data()
        self.plan_list_changed.emit()
        self.current_plan_changed.emit(self.domain_service.get_current_index())
        self.point_list_changed.emit()
        return True
    return False

switch_plan

switch_plan(index: int)

切换方案。

参数：

名称	类型	描述	默认
index	int	目标方案索引。	必需

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def switch_plan(self, index: int):
    """切换方案。

    Args:
        index (int): 目标方案索引。
    """
    if self.domain_service.set_current_index(index):
        self._save_data()
        self.current_plan_changed.emit(index)
        self.point_list_changed.emit()

rename_plan

rename_plan(index: int, new_name: str)

重命名方案。

参数：

名称	类型	描述	默认
index	int	方案索引。	必需
new_name	str	新名称。	必需

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def rename_plan(self, index: int, new_name: str):
    """重命名方案。

    Args:
        index (int): 方案索引。
        new_name (str): 新名称。
    """
    if self.domain_service.rename_plan(index, new_name):
        self._save_data()
        self.plan_list_changed.emit()

add_point

add_point(point_data: dict)

添加节点

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def add_point(self, point_data: dict):
    """添加节点"""
    self.domain_service.add_point(point_data)
    self._save_data()
    self.point_list_changed.emit()

delete_point

delete_point(index: int)

删除节点

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def delete_point(self, index: int):
    """删除节点"""
    self.domain_service.remove_point(index)
    self._save_data()
    self.point_list_changed.emit()

move_point_up

move_point_up(index: int)

上移节点

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def move_point_up(self, index: int):
    """上移节点"""
    if self.domain_service.move_point_up(index):
        self._save_data()
        self.point_list_changed.emit()

move_point_down

move_point_down(index: int)

下移节点

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def move_point_down(self, index: int):
    """下移节点"""
    if self.domain_service.move_point_down(index):
        self._save_data()
        self.point_list_changed.emit()

update_point

update_point(index: int, point_data: dict)

更新节点

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def update_point(self, index: int, point_data: dict):
    """更新节点"""
    self.domain_service.update_point(index, point_data)
    self._save_data()
    self.point_list_changed.emit()

get_single_point

get_single_point(index: int) -> dict

获取单个节点数据。

参数：

名称	类型	描述	默认
index	int	节点索引。	必需

返回：

名称	类型	描述
dict	dict	节点数据字典，如果索引无效则返回None。

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def get_single_point(self, index: int) -> dict:
    """获取单个节点数据。

    Args:
        index (int): 节点索引。

    Returns:
        dict: 节点数据字典，如果索引无效则返回None。
    """
    points = self.domain_service.get_all_points()
    if 0 <= index < len(points):
        return points[index]
    return None

execute_single_point

execute_single_point(index: int)

执行单个节点。

流程： 1. 获取节点数据 2. 解析为任务列表 3. 准备执行操作 4. 查询当前位置

参数：

名称	类型	描述	默认
index	int	节点索引。	必需

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def execute_single_point(self, index: int):
    """执行单个节点。

    流程：
    1. 获取节点数据
    2. 解析为任务列表
    3. 准备执行操作
    4. 查询当前位置

    Args:
        index (int): 节点索引。
    """
    tasks = self._get_node_tasks(index)
    if not tasks:
        return

    # 准备执行操作
    self.motion_constructor.prepare_operation(
        MotionOperationMode.EXECUTE,
        tasks
    )

    # 查询当前位置，触发执行
    self.command_hub.get_current_position()

execute_motion_plan

execute_motion_plan()

执行整个运动规划方案。

流程： 1. 获取所有节点的任务 2. 准备执行操作 3. 查询当前位置，触发执行

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def execute_motion_plan(self):
    """执行整个运动规划方案。

    流程：
    1. 获取所有节点的任务
    2. 准备执行操作
    3. 查询当前位置，触发执行
    """
    tasks = self._get_plan_tasks()
    if not tasks:
        return

    # 准备执行操作
    self.motion_constructor.prepare_operation(
        MotionOperationMode.EXECUTE,
        tasks
    )

    # 查询当前位置，触发执行
    self.command_hub.run_motion_sequence()

request_current_position

request_current_position()

请求获取当前位置（仅用于UI显示）。

流程： 1. 发送 0x07 命令查询位置 2. 串口返回数据后，MessageResponseService 检查 has_pending_operation() 3. 因为没有调用 prepare_operation，所以返回 False 4. MessageResponseService emit get_current_position_signal 5. 本Service转发信号给ViewModel

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def request_current_position(self):
    """请求获取当前位置（仅用于UI显示）。

    流程：
    1. 发送 0x07 命令查询位置
    2. 串口返回数据后，MessageResponseService 检查 has_pending_operation()
    3. 因为没有调用 prepare_operation，所以返回 False
    4. MessageResponseService emit get_current_position_signal
    5. 本Service转发信号给ViewModel
    """
    self.command_hub.single_send_command(control=0x07)

save_node_trajectory

save_node_trajectory(node_index: int) -> bool

保存单个节点的轨迹。

流程： 1. 获取节点任务 2. 准备保存操作 3. 查询当前位置 4. MessageResponseService 自动完成保存

参数：

名称	类型	描述	默认
node_index	int	节点索引。	必需

返回：

名称	类型	描述
bool	bool	True: 准备成功, False: 准备失败。

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def save_node_trajectory(self, node_index: int) -> bool:
    """保存单个节点的轨迹。

    流程：
    1. 获取节点任务
    2. 准备保存操作
    3. 查询当前位置
    4. MessageResponseService 自动完成保存

    Args:
        node_index (int): 节点索引。

    Returns:
        bool: True: 准备成功, False: 准备失败。
    """
    try:
        tasks = self._get_node_tasks(node_index)
        if not tasks:
            return False

        current_plan = self.domain_service.get_current_plan()
        if not current_plan:
            return False

        context = {
            "filename": f"{current_plan.name}-{node_index}",
            "type": "node"
        }

        self.motion_constructor.prepare_operation(
            MotionOperationMode.SAVE,
            tasks,
            context
        )

        self.command_hub.get_current_position()
        return True
    except Exception:
        return False

save_plan_trajectory

save_plan_trajectory() -> bool

保存整个方案的轨迹

流程： 1. 获取方案所有任务 2. 准备保存操作 3. 查询当前位置 4. MessageResponseService 自动完成保存

返回：

名称	类型	描述
True	bool	准备成功
False	bool	准备失败

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def save_plan_trajectory(self) -> bool:
    """
    保存整个方案的轨迹

    流程：
    1. 获取方案所有任务
    2. 准备保存操作
    3. 查询当前位置
    4. MessageResponseService 自动完成保存

    Returns:
        True: 准备成功
        False: 准备失败
    """
    try:
        tasks = self._get_plan_tasks()
        if not tasks:
            return False

        current_plan = self.domain_service.get_current_plan()
        if not current_plan:
            return False

        context = {
            "filename": current_plan.name,
            "type": "plan"
        }

        self.motion_constructor.prepare_operation(
            MotionOperationMode.SAVE,
            tasks,
            context
        )

        self.command_hub.get_current_position()
        return True
    except Exception:
        return False

preview_node_trajectory

preview_node_trajectory(node_index: int) -> bool

预览单个节点的轨迹曲线

流程： 1. 获取节点任务 2. 准备预览操作 3. 查询当前位置 4. MessageResponseService 自动发射预览信号

参数：

名称	类型	描述	默认
node_index	int	节点索引	必需

返回：

名称	类型	描述
True	bool	准备成功
False	bool	准备失败

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def preview_node_trajectory(self, node_index: int) -> bool:
    """
    预览单个节点的轨迹曲线

    流程：
    1. 获取节点任务
    2. 准备预览操作
    3. 查询当前位置
    4. MessageResponseService 自动发射预览信号

    Args:
        node_index: 节点索引

    Returns:
        True: 准备成功
        False: 准备失败
    """
    try:
        tasks = self._get_node_tasks(node_index)
        if not tasks:
            return False

        context = {
            "type": "node",
            "node_index": node_index
        }

        self.motion_constructor.prepare_operation(
            MotionOperationMode.PREVIEW,
            tasks,
            context
        )

        self.command_hub.get_current_position()
        return True
    except Exception:
        return False

preview_plan_trajectory

preview_plan_trajectory() -> bool

预览整个方案的轨迹曲线

流程： 1. 获取方案所有任务 2. 准备预览操作 3. 查询当前位置 4. MessageResponseService 自动发射预览信号

返回：

名称	类型	描述
True	bool	准备成功
False	bool	准备失败

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def preview_plan_trajectory(self) -> bool:
    """
    预览整个方案的轨迹曲线

    流程：
    1. 获取方案所有任务
    2. 准备预览操作
    3. 查询当前位置
    4. MessageResponseService 自动发射预览信号

    Returns:
        True: 准备成功
        False: 准备失败
    """
    try:
        tasks = self._get_plan_tasks()
        if not tasks:
            return False

        context = {"type": "plan"}

        self.motion_constructor.prepare_operation(
            MotionOperationMode.PREVIEW,
            tasks,
            context
        )

        self.command_hub.get_current_position()
        return True
    except Exception:
        return False

load_local_trajectory

load_local_trajectory(file_path: str) -> bool

从 plans 目录加载轨迹文件，作为示教节点添加

参数：

名称	类型	描述	默认
file_path	str	轨迹文件的完整路径或文件名	必需

返回：

名称	类型	描述
True	bool	加载成功
False	bool	加载失败

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def load_local_trajectory(self, file_path: str) -> bool:
    """
    从 plans 目录加载轨迹文件，作为示教节点添加

    Args:
        file_path: 轨迹文件的完整路径或文件名

    Returns:
        True: 加载成功
        False: 加载失败
    """
    try:
        # 1. 提取文件名（不带路径和扩展名）
        file_stem = Path(file_path).stem

        # 2. 从 TrajectoryRepository 加载
        trajectory_data = self.trajectory_repository.load_trajectory(file_stem)

        # 3. 验证数据
        if not self._validate_trajectory_data(trajectory_data):
            return False

        # 4. 创建示教节点数据（复用示教模式的格式）
        point_data = {
            "mode": f"示教-{file_stem}",
            "joint1": 0.0,
            "joint2": 0.0,
            "joint3": 0.0,
            "joint4": 0.0,
            "joint5": 0.0,
            "joint6": 0.0,
            "frequency": 0.01,
            "curve_type": "S曲线",
            "gripper_command": "00: 不进行任何操作",
            "gripper_param": 0.0,
            "note": f"本地轨迹，共{len(trajectory_data)}个点",
            "teach_record_name": file_stem,
            "teach_data": trajectory_data,
            "source": "local_trajectory"  # 标记来源
        }

        # 5. 添加节点
        self.domain_service.add_point(point_data)
        self._save_data()  # 持久化保存
        self.point_list_changed.emit()
        return True

    except Exception:
        return False

get_local_trajectory_files

get_local_trajectory_files() -> List[str]

获取所有可用的本地轨迹文件

返回：

类型	描述
List[str]	文件名列表（不带扩展名）

源代码位于： src/controller/controller/application/services/motion_planning_application_service.py

def get_local_trajectory_files(self) -> List[str]:
    """
    获取所有可用的本地轨迹文件

    Returns:
        文件名列表（不带扩展名）
    """
    return self.trajectory_repository.list_trajectory_files()