202505 海康机器人客制叉车

叉车可以举升，门架可以前移。控制举升和门架都是通过脚本。

叉车进行两层货架与地面之间的相互搬运。

脚本动作

叉子动作

在车模型中关于叉子得硬件信息如下：

• lift_pump: 泵
• lift_up： 叉子抬升
• lift_down ： 叉子下降
• wireEncoder：拉线编码器

抬升叉子时，需要抬升与泵IO同时设置

下降叉子时，需要抬升与下降IO同时设置

拉线编码器获取当时的叉子高度。

叉子脚本如下：

execute_with_value_timeout 做了叉子动作的超时，如果2s拉线编码器值未发变化。则任务异常。

check_func 函数检擦每次拉线编码器值得时候，同时检测有没有手动，急停得异常情况。

信息

获取拉线编码器的值的时候，由于没有校准，所以需要一个获取值与实际高度的一个转化。

from abc import abstractmethod
from actuator import Actuator, TimeoutResult
from sec_dev import IoApi, BusinessStatus, SecurityAlarmApi
from dev_api import SecDevInterface
import logging

class Forklift(Actuator):
    """叉车执行器类，继承自Actuator抽象基类"""
    
    def __init__(self, logger: logging.Logger = None):
        super().__init__(logger)
        # 定义IO点位的名称
        self.PUMP_DO_NAME = "lift_pump"
        self.LIFT_UP_DO_NAME = "lift_up"
        self.LIFT_DOWN_DO_NAME = "lift_down"
        self.WIRE_ENCODER_NAME = "wireEncoder"
        
    def check_emergency(self,robot: SecDevInterface = None) -> bool:
        """检查紧急情况
        
        Returns:
            bool: 如果存在紧急情况返回True，否则返回False
        """
        alarm_dict = SecurityAlarmApi.get_all_alarm_info(robot)
        alarm_info:dict = alarm_dict.get("info", {})
        # 遍历字典，检测碰撞条或急停是否触发
        for alarm in alarm_info:
            if alarm["m_type"] == 3 or alarm["m_type"] == 4 or alarm["m_type"] == 5 or alarm["m_type"] == 7:
                self.log.info(f"急停或者碰撞条触发 {alarm_info}")
                return True
        return False
    
    def stop(self, robot: SecDevInterface = None):
        """停止叉车所有操作"""
        if robot is None:
            self.log.error("robot参数为None，无法执行停止操作")
            return
        
        try:
            self.set_do_value(robot, self.PUMP_DO_NAME, 0)
            self.set_do_value(robot, self.LIFT_UP_DO_NAME, 0)
            self.set_do_value(robot, self.LIFT_DOWN_DO_NAME, 0)
            self.log.info("紧急停止完成")
        except Exception as e:
            self.log.error(f"紧急停止失败: {e}")
    
    def check_status(self, robot: SecDevInterface):
        """检查叉车状态
        
        Args:
            robot: 机器人接口实例
            
        """
        try:
            # 获取油泵电机DO状态
            pump_status = IoApi.get_di(robot, {"di_name": self.PUMP_DO_NAME})
            
            # 获取上升电磁阀DO状态
            lift_up_status = IoApi.get_di(robot, {"di_name": self.LIFT_UP_DO_NAME})
            
            # 获取下降电磁阀DO状态
            lift_down_status = IoApi.get_di(robot, {"di_name": self.LIFT_DOWN_DO_NAME})
            
            # 获取货叉高度
            fork_height = self.get_line(robot)
            
            # 记录状态到日志
            self.log.info(f"叉车状态检查 -> "
                         f"油泵电机({self.PUMP_DO_NAME}): {pump_status}, "
                         f"上升阀({self.LIFT_UP_DO_NAME}): {lift_up_status}, "
                         f"下降阀({self.LIFT_DOWN_DO_NAME}): {lift_down_status}, "
                         f"货叉高度: {fork_height}")
            
        except Exception as e:
            self.log.error(f"检查叉车状态失败: {e}")
      
    def get_line(self, robot: SecDevInterface):
        """获取校正后的拉线值"""
        line_code = IoApi.get_line_code(robot, {"name": self.WIRE_ENCODER_NAME})
        self.log.debug(f"拉线真实值: {line_code}")
        corrected_value = -2 * line_code + 44
        return corrected_value

    def lift(self, robot: SecDevInterface, target_height):
        """叉子抬升控制函数"""
        line_code = self.get_line(robot)
        self.log.debug(f"下发举升高度: {target_height}, 当前拉线: {line_code}")
        
        if line_code < 0:
            raise Exception("获取拉线值不能为负数")
            
        if target_height == line_code:
            self.log.info("目标高度已达到，无需移动")
            return
            
        going_up = target_height > line_code
        
        def action_func(**kwargs):
            if going_up:
                self.log.info(f"开始举升: 从 {line_code} 到 {target_height}")
                self.set_do_value(robot, self.LIFT_DOWN_DO_NAME, 0)  # 先确保下降阀关闭
                self.set_do_value(robot, self.LIFT_UP_DO_NAME, 1)
                self.set_do_value(robot, self.PUMP_DO_NAME, 1)
            else:
                self.log.info(f"开始下放: 从 {line_code} 到 {target_height}")
                self.set_do_value(robot, self.LIFT_UP_DO_NAME, 1)
                self.set_do_value(robot, self.LIFT_DOWN_DO_NAME, 1)
            return True  # 表示动作已下发
                
        def check_func(**kwargs):

            if self.check_emergency(robot):
                raise Exception("急停或者碰撞条触发")

            return self.get_line(robot)
                
        # 定义目标检查函数
        def target_check_func(current_pos):
            self.log.info(f"检查当前值{current_pos} 目标值{target_height}")
            if going_up:  # 举升
                return current_pos >= target_height
            else:  # 下放
                return abs(current_pos - target_height) <= 2
                
        def cleanup_func(**kwargs):
            self.log.info("执行清理：关闭泵与电磁阀")
            self.stop(robot)
            
        result, last_value, elapsed = self.execute_with_value_timeout(
            action_func=action_func,
            check_func=check_func,
            check_interval=0.05,
            value_timeout=2,
            target_value=target_check_func,
            cleanup_func=cleanup_func,
            robot=robot,
            target_height=target_height,
            going_up=going_up
        )
        
        # 检查执行结果，如果超时则抛出异常

        if result in [TimeoutResult.TIMEOUT, TimeoutResult.VALUE_TIMEOUT]:
            raise Exception(f"叉车升降操作超时: {result.value}, 目标高度={target_height}, 最后位置={last_value}, 耗时={elapsed:.2f}秒")
        elif result == TimeoutResult.FAILED:
            raise Exception(f"叉车升降操作失败: 目标高度={target_height}, 最后位置={last_value}, 耗时={elapsed:.2f}秒")
        
        self.stop(robot)

门架动作

在车模型中关于门架得硬件信息如下：

• lift_pump: 泵
• sway_go: 门架推出
• sway_back: 门架收回
• sway_zero: 门架零位
• sway_final: 门架推出到位

推出时，需要门架推出并开泵。并检测门架推出到位。 收回时，需要门前收回并开泵。并检测门架收回到位。

门架脚本如下：

execute_with_timeout 执行推出，收回门架时，设置了10秒得延时时间， 如果超时则异常。

check_func 每次获取对应到位信号得同时，也检测是否手动或者急停得状态。

from abc import abstractmethod
from actuator import Actuator, TimeoutResult
from sec_dev import IoApi, BusinessStatus, SecurityAlarmApi
from dev_api import SecDevInterface
import logging

class Sway(Actuator):
    """门架执行器类，继承自Actuator抽象基类"""
    
    def __init__(self, logger: logging.Logger = None):
        super().__init__(logger)
        # 定义IO点位的名称
        self.PUMP_DO_NAME = "lift_pump"
        self.SWAY_GO_DO_NAME = "sway_go"      # 门架伸出
        self.SWAY_BACK_DO_NAME = "sway_back"  # 门架收缩
        self.SWAY_ZERO_DI_NAME = "sway_zero"  # 门架零位检测
        self.SWAY_FINAL_DI_NAME = "sway_final" # 门架最终位检测
        
    def check_emergency(self, robot: SecDevInterface = None) -> bool:
        """检查紧急情况
        
        Returns:
            bool: 如果存在紧急情况返回True，否则返回False
        """
        alarm_dict = SecurityAlarmApi.get_all_alarm_info(robot)
        alarm_info:dict = alarm_dict.get("info", {})
        # 遍历字典，检测碰撞条或急停是否触发
        for alarm in alarm_info:
            if alarm["m_type"] == 3 or alarm["m_type"] == 4 or alarm["m_type"] == 5 or alarm["m_type"] == 7:
                self.log.info(f"急停或者碰撞条触发 {alarm_info}")
                return True
        return False
    
    def stop(self, robot: SecDevInterface = None):
        """停止门架所有操作"""
        if robot is None:
            self.log.error("robot参数为None，无法执行停止操作")
            return
        
        try:
            self.set_do_value(robot, self.PUMP_DO_NAME, 0)
            self.set_do_value(robot, self.SWAY_GO_DO_NAME, 0)
            self.set_do_value(robot, self.SWAY_BACK_DO_NAME, 0)
            self.log.info("门架紧急停止完成")
        except Exception as e:
            self.log.error(f"门架紧急停止失败: {e}")
    
    def check_status(self, robot: SecDevInterface):
        """检查门架状态
        
        Args:
            robot: 机器人接口实例
            
        """
        try:
            # 获取油泵电机DO状态
            pump_status = IoApi.get_di(robot, {"di_name": self.PUMP_DO_NAME})
            
            # 获取门架伸出电磁阀DO状态
            sway_go_status = IoApi.get_di(robot, {"di_name": self.SWAY_GO_DO_NAME})
            
            # 获取门架收缩电磁阀DO状态
            sway_back_status = IoApi.get_di(robot, {"di_name": self.SWAY_BACK_DO_NAME})
            
            # 获取门架位置状态
            zero_detected = self.get_limit_di(robot, self.SWAY_ZERO_DI_NAME)
            final_detected = self.get_limit_di(robot, self.SWAY_FINAL_DI_NAME)
            
            # 记录状态到日志
            self.log.info(f"门架状态检查 -> "
                         f"油泵电机({self.PUMP_DO_NAME}): {pump_status}, "
                         f"伸出阀({self.SWAY_GO_DO_NAME}): {sway_go_status}, "
                         f"收缩阀({self.SWAY_BACK_DO_NAME}): {sway_back_status}, "
                         f"零位检测: {zero_detected}, "
                         f"最终位检测: {final_detected}")
            
        except Exception as e:
            self.log.error(f"检查门架状态失败: {e}")
    
    def get_limit_di(self, robot: SecDevInterface, name):
        """获取限位DI状态"""
        di_val = IoApi.get_di(robot, {"di_name": str(name)})
        self.log.debug(f"限位IO: {name}, di_val {di_val}")
        return di_val

    def sway(self, robot: SecDevInterface, target):
        """门架前移后移控制函数
        
        Args:
            robot: 机器人接口实例
            target: 目标位置，1表示伸出，0表示收缩
        """
        self.log.debug(f"门架移动目标: {target}")
        
        if target == 1:  # 伸出
            
            def action_func(**kwargs):
                self.log.info("开始门架伸出")
                self.set_do_value(robot, self.SWAY_BACK_DO_NAME, 0)  # 先确保收缩阀关闭
                self.set_do_value(robot, self.SWAY_GO_DO_NAME, 1)
                self.set_do_value(robot, self.PUMP_DO_NAME, 1)
                return True
                
            def check_func(**kwargs):
                if self.check_emergency(robot):
                    raise Exception("急停或者碰撞条触发")
                final_detected = self.get_limit_di(robot, self.SWAY_FINAL_DI_NAME)
                return final_detected
                
            def target_check_func(current_status):
                # 当检测到货物或到达最终位时停止
                return current_status == 0
                
            def cleanup_func(**kwargs):
                self.log.info("开始门架伸出清理")
                self.stop(robot)
                    
        elif target == 0:  # 收缩

            def action_func(**kwargs):
                self.log.info("开始门架收缩")
                self.set_do_value(robot, self.SWAY_GO_DO_NAME, 0)  # 先确保伸出阀关闭
                self.set_do_value(robot, self.SWAY_BACK_DO_NAME, 1)
                self.set_do_value(robot, self.PUMP_DO_NAME, 1)
                return True
                
            def check_func(**kwargs):
                if self.check_emergency(robot):
                    raise Exception("急停或者碰撞条触发")
                # 返回零位检测状态
                return self.get_limit_di(robot, self.SWAY_ZERO_DI_NAME)
                
            def target_check_func(zero_detected):
                return zero_detected == 0
                
            def cleanup_func(**kwargs):
                self.log.info("开始门架收缩清理")
                self.stop(robot)

        else:
            raise Exception(f"无效的目标值: {target}")
            
        # 执行超时控制
        result, last_value, elapsed = self.execute_with_timeout(
            action_func=action_func,
            check_func=check_func,
            timeout=10,
            check_interval=0.05,
            value_timeout=0,
            target_value=target_check_func,
            max_retries=1,
            retry_interval=0.5,
            cleanup_func=cleanup_func,
            robot=robot,
            target=target
        )
        
        # 检查执行结果，如果超时则抛出异常
        if result in [TimeoutResult.TIMEOUT, TimeoutResult.VALUE_TIMEOUT]:
            action_desc = "伸出" if target == 1 else "收缩"
            raise Exception(f"门架{action_desc}操作超时: {result.value}, 目标={target}, 最后状态={last_value}, 耗时={elapsed:.2f}秒")
        elif result == TimeoutResult.FAILED:
            action_desc = "伸出" if target == 1 else "收缩"
            raise Exception(f"门架{action_desc}操作失败: 目标={target}, 最后状态={last_value}, 耗时={elapsed:.2f}秒")
        
        # 确保停止所有动作
        self.stop(robot)

    def extend(self, robot: SecDevInterface):
        """门架伸出操作"""
        self.sway(robot, 1)
    
    def retract(self, robot: SecDevInterface):
        """门架收缩操作"""
        self.sway(robot, 0)

最终执行脚本

最终需要执行取货放货，需要同时执行叉子和门架。我们将前面两个脚本中的函数组合实现搬运动作。

搬运动作脚本如下：

在 steps 中执行 step() 中需要每次延时1s。这样看起来更安全一些。

import datetime
import time
import logging
from forklift import Forklift
from sway import Sway
from sec_dev import TemplateBusiness, BusinessStatus
from dev_api import SecDevInterface
import utils

# 创建日志输出文件
log = logging.getLogger("forklift_pick_place")
 
log.handlers.clear()
if not log.handlers:
    file_handler = logging.FileHandler('./forklift_pick_place.log', encoding="utf8")
    log.setLevel(logging.DEBUG)
 
    fmt = '''[%(asctime)s]-%(name)s-%(levelname)s:''' \
          '''%(message)s-%(filename)s:%(lineno)s'''
    formatter = logging.Formatter(fmt)
    file_handler.setFormatter(formatter)
    file_handler.setLevel(logging.DEBUG)
    log.addHandler(file_handler)

log.info("叉车取放货业务脚本启动......")

class Business(TemplateBusiness):
    """叉车取放货业务类，继承自TemplateBusiness"""
    
    def __init__(self, robot: SecDevInterface, params):
        super().__init__()
        
        log.debug(f"参数: {params}")
        
        # 设置默认参数
        if params == "":
            self.operation_param = {
                "action": "pick",  # pick: 取货, place: 放货
                "target_height": 2000,  # 目标高度
                "lift_offset": 100      # 抬升高度
            }
        else:
            self.operation_param = params

        # 初始化叉车和门架执行器
        self.forklift = Forklift(log)
        self.sway = Sway(log)
        
        # 清除模块告警
        utils.set_module_alarm(robot, 5, 0, "", "")

    def execute(self, robot: SecDevInterface, args):
        """执行叉车取放货操作"""
        
        action = self.operation_param.get("action")
        target_height = int(self.operation_param.get("target_height"))
        lift_offset = int(self.operation_param.get("lift_offset"))
        
        log.info(f"执行操作: {action}, 目标高度: {target_height}, 抬升高度: {lift_offset}")
        
        if action == "pick":
            # 取货流程：
            # 1. 降叉子到目标高度-抬升高度
            # 2. 推门架
            # 3. 抬升到目标高度
            # 4. 收门架
            initial_height = target_height - lift_offset
            steps = [
                lambda: self.forklift.lift(robot, initial_height),  # 降叉子到初始高度
                lambda: self.sway.extend(robot),  # 推门架（伸出）
                lambda: self.forklift.lift(robot, target_height),  # 抬升到目标高度
                lambda: self.sway.retract(robot)  # 收门架（收缩）
            ]
        elif action == "place":
            # 放货流程：
            # 1. 抬升到目标高度+抬升高度
            # 2. 推门架
            # 3. 降到目标高度
            # 4. 收门架
            elevated_height = target_height + lift_offset
            steps = [
                lambda: self.forklift.lift(robot, elevated_height),  # 抬升到抬高位置
                lambda: self.sway.extend(robot),  # 推门架（伸出）
                lambda: self.forklift.lift(robot, target_height),  # 降到目标高度
                lambda: self.sway.retract(robot)  # 收门架（收缩）
            ]
        else:
            log.error(f"不支持的操作类型: {action}")
            utils.set_module_alarm(robot, 5, 1, f"不支持的操作类型 {action}", "")
            return BusinessStatus.FAILED

        for i, step in enumerate(steps):
            try:
                step()
                time.sleep(1)  # 每步操作后等待1秒
                self.status = BusinessStatus.FINISHED
            except Exception as e:
                log.error(f"步骤 {i} 执行失败: {e}")
                utils.set_module_alarm(robot, 6, 1, f"叉车{action}任务第{i}步执行失败", "")
                self.status = BusinessStatus.FAILED
                return self.status

        return BusinessStatus.FINISHED

编写任务编排

先空车移动到起点，执行取货动作。再空车移动到终点，执行放货动作。

可以参考移栽的搬运。

空车移动（锁车）-> 抬升叉子（锁车）-> 空车移动（锁车）-> 降低叉子（锁车）

锁车功能说明：

前一个任务把车锁定了，下个任务就必须指定这辆车，所以后面的任务，它车号的数据来源是来自于前一个任务

充电

叉车使用的地盘是潜伏车，车头充电。 实际设置了站点终点方向180度之后，小车还是叉子朝充电桩。

需要此时需要配置充电桩关联站点。

提示

在添加充电外设的时候，需要鼠标点击拖动滚动条来选择充电桩。如果使用滚动条则会跳过充桩。

PDA

添加项目

路径：系统管理->添加

添加服务

添加Rcs服务，服务路径为：https://192.168.124.120/

增加关联接口标识： logout

方法：Post

URL： /unify/basic/api/unify/logout

添加应用

选择应用PDA， ClientView 和 RCS 配合使用，选择PDA（标准版）

添加扩展脚本

this.onLogin = async (info) => {
  try {
    // 校验参数
    if (!info.username || !info.password) {
      throw {
        message: this.$t('login.error.need')
      }
    }
    const param = {
      username: info.username,
      password: sha256(info.password), // 需要对密码做处理
      clientType: 'pda'
    }
    // 调用登录接口
    const { data } = await this.$post('/unify/basic/api/unify/token', param)
    if (data.success) {
      // 更新token
      mutations.updateToken(data)
      // 登录成功之后跳转至首页
      this.$router.push('/home')
    } else {
      throw data
    }
  } catch(e) {
    this.loginTip = e.message || this.$t('login.error.unknown')
  }
}

上下文

属性名	属性值	类型
LOGOUT_METHOD	RCS.logout	String

配置页面

新建PDA通用版本，页面配置中增加一个一层取放货的页面。

该页面脚本编写以下提交任务的脚本。

起点站点控件的id设置为：#startCode

终点站点控件的id 设置为：#endCode

提交的Button的事件中绑定： onSubmit。

// 提交按钮
this.onSubmit = () => {
    const params = {
        targetRoute: [{
                type: 'SITE',
                code: this.form['#startCode'],
                seq: 0
            },
            {
                type: 'SITE',
                code: this.form['#endCode'],
                seq: 1
            }
        ],
        taskType: 'L01'
    }
    const priority = this.form['#initPriority']
    if (!priority || priority === '') {
        params.initPriority = 1
    } else {
        params.initPriority = parseInt(priority)
    }
    const headers = {'X-lr-request-id': new Date().getTime()}
    this.$post('https://192.168.124.120/rcs/rtas/web/schedule/submit', params, {headers}).then((res) => {
        if (res.data.success || res.data.code === '0') {
            this.$tip(this.$text('下发成功'), 'success')
            this.clear()
        } else {
            this.$tip(res.data.message)
        }
    })
}

发布

获取应用id和项目id，拼接一个页面url如下：

https://192.168.124.120/clientview/apps/base-pda/index.html?appId=20250812153631898&projectId=20250812140847670#/login

在浏览器访问该地址如果能访问到说明Url拼接成功。

将前面ip地址部分拿掉 /clientview/apps/base-pda/index.html?appId=20250812153631898&projectId=20250812140847670#/login 配置到PDA登录页面右上角的加载路径中。