27 KiB
VWED在线脚本内置函数文档
本文档包含VWED在线脚本系统中所有可用的内置函数,按功能模块分类整理,为用户提供开发指导。
目录
协议通信
Fins协议
read_fins_string
描述: Fins读取字符串
def read_fins_string(ip: str, port: int, area: int, fins_io_addr: int, bit_offset: int, word_length: int) -> str | None
参数:
ip: str类型,从机IPport: int类型,从机端口area: int类型,读取的存储区域代码,16进制值如:0x82fins_io_addr: int类型,读取的地址位bit_offset: int类型,读取的偏移量word_length: int类型,读取的长度
返回值: 成功返回字符串,失败返回None
异常: 本方法不抛出异常,异常捕获后只作日志记录
read_fins_word
描述: Fins读取word
def read_fins_word(ip: str, port: int, area: int, fins_io_addr: int, bit_offset: int) -> int | None
参数:
ip: str类型,从机IPport: int类型,从机端口area: int类型,读取的存储区域代码,16进制值如:0x82fins_io_addr: int类型,读取的地址位bit_offset: int类型,读取的偏移量
返回值: 成功返回整数,失败返回None
read_fins_bit
描述: Fins读取bit
def read_fins_bit(ip: str, port: int, area: int, fins_io_addr: int, bit_offset: int) -> int | None
参数: 同read_fins_word
返回值: 成功返回整数,失败返回None
write_fins_word
描述: Fins写入word
def write_fins_word(ip: str, port: int, area: int, fins_io_addr: int, bit_offset: int, value: int) -> None
参数: 在read_fins_word基础上增加value参数(写入的值)
write_fins_bit
描述: Fins写入bit
def write_fins_bit(ip: str, port: int, area: int, fins_io_addr: int, bit_offset: int, value: bool) -> None
参数: 在read_fins_bit基础上增加value参数(bool类型,写入的值)
HTTP请求
request_post
描述: 发送POST请求,参数为JSON格式
def request_post(url: str, param: str) -> str | None
参数:
url: str类型,请求的URLparam: str类型,JSON字符串,请求的参数
返回值: 成功返回响应JSON字符串,失败返回None
request_get
描述: 发送GET请求
def request_get(url: str) -> str | None
参数:
url: str类型,请求的URL
返回值: 成功返回请求返回的字符串,失败返回None
set_header
描述: 设置请求头,设置一次后对所有请求都有效
def set_header(key: str, value: str) -> None
参数:
key: str类型,请求头的keyvalue: str类型,请求头的value
request_post_xml
描述: 发送POST请求,参数为XML格式
def request_post_xml(url: str, param: str) -> str | None
参数:
url: str类型,请求的URLparam: str类型,JSON字符串,会自动转换为XML格式
返回值: 成功时将响应的XML格式字符串转化为JSON格式并返回,失败返回None
request_http_post
描述: 发送POST请求(增强版),可设置请求媒体类型和请求头
def request_http_post(url: str, param: str, head_param: str, media_type: str) -> str | None
参数:
url: 请求的URLparam: 请求的入参json字符串head_param: 请求头,如有传json字符串,无传空字符串media_type: 请求类型,可选值:JSON、JAVASCRIPT、HTML、XML、XWWWFORMURLENCODED
返回值: 成功返回json字符串,失败返回None
MQTT协议
mqtt_subscribe
描述: MQTT订阅信息(使用传入的topic)
def mqtt_subscribe(topic: str) -> str
参数:
topic: str类型,订阅的topic
返回值: 订阅到的信息,失败返回None
使用示例:
data = mqtt_subscribe("test_topic")
script_log("info", "sub1", data)
mqtt_publish
描述: MQTT发布信息(使用传入的topic)
def mqtt_publish(topic: str, message: str) -> None
参数:
topic: str类型,发布的topicmessage: str类型,需要发布的信息
使用示例:
mqtt_publish("test_topic", "RDS-Script sub message.")
Melsec协议
read_melsec_boolean
描述: Melsec读取boolean
def read_melsec_boolean(ip: str, port: int, address: str) -> bool | None
参数:
ip: str类型,从机IPport: int类型,从机端口address: str类型,读取的地址位
返回值: 成功返回布尔值,失败返回None
read_melsec_number
描述: Melsec读取number
def read_melsec_number(ip: str, port: int, address: str) -> float | None
参数: 同read_melsec_boolean
返回值: 成功返回浮点数,失败返回None
read_melsec_string
描述: Melsec读取string
def read_melsec_string(ip: str, port: int, address: str, length: int) -> str | None
参数: 在read_melsec_boolean基础上增加length参数(读取的长度)
返回值: 成功返回字符串,失败返回None
write_melsec_boolean
描述: Melsec写入boolean
def write_melsec_boolean(ip: str, port: int, address: str, value: bool) -> bool
参数: 在read_melsec_boolean基础上增加value参数
返回值: 成功返回True,失败返回False
write_melsec_number
描述: Melsec写入number
def write_melsec_number(ip: str, port: int, address: str, value: float) -> bool
write_melsec_string
描述: Melsec写入string
def write_melsec_string(ip: str, port: int, address: str, value: str) -> bool
Modbus TCP协议
read_coil_status
描述: Modbus读取线圈量,功能码01
def read_coil_status(ip: str, port: int, slave_id: int, offset: int) -> bool | None
参数:
ip: str类型,从机IPport: int类型,从机端口slave_id: int类型,从机slave IDoffset: int类型,Modbus地址
返回值: 成功返回布尔值,失败返回None
read_input_status
描述: Modbus读取离散输入,功能码02
def read_input_status(ip: str, port: int, slave_id: int, offset: int) -> bool | None
read_holding_register
描述: Modbus读取保持寄存器,功能码03
def read_holding_register(ip: str, port: int, slave_id: int, offset: int, data_type: int) -> int | None
参数: 在基础参数基础上增加data_type(2:无符号整形,3:有符号整形)
read_input_register
描述: Modbus读取输入寄存器,功能码04
def read_input_register(ip: str, port: int, slave_id: int, offset: int, data_type: int) -> int | None
write_coil_status
描述: Modbus写入线圈量,功能码05
def write_coil_status(ip: str, port: int, slave_id: int, offset: int, value: bool) -> bool
返回值: True:写入成功,False:写入失败
write_holding_register
描述: Modbus写入保持寄存器,功能码06
def write_holding_register(ip: str, port: int, slave_id: int, offset: int, data_type: int, value: int) -> bool
批量操作函数
batch_read_coil_status
描述: Modbus批量读取线圈量,功能码01
def batch_read_coil_status(ip: str, port: int, slave_id: int, offset: int, len: int) -> list[bool] | None
参数: 增加len参数(Modbus地址数量)
返回值: 成功返回布尔值数组,失败返回None
batch_read_holding_registers
描述: Modbus批量读取保持寄存器,功能码03
def batch_read_holding_registers(ip: str, port: int, slave_id: int, offset: int, len: int) -> list[int] | None
batch_write_coil_status
描述: Modbus批量写入线圈量,功能码0f
def batch_write_coil_status(ip: str, port: int, slave_id: int, offset: int, value: list[bool]) -> bool
通用操作函数
read_single_modbus_value
描述: 通用读取单个地址
def read_single_modbus_value(ip: str, port: int, slave_id: int, type_str: str, offset: int) -> int | None
参数: type_str可选择'0x','1x','3x','4x'
read_single_modbus_value_by_instance_name
描述: 通过实例名称读取单个地址
def read_single_modbus_value_by_instance_name(name: str, offset: int) -> int | None
参数:
name: str类型,配置的唯一实例名称offset: int类型,Modbus地址
OPC UA协议
read_opc_value
描述: 读取OPC server中的某个地址位的值
def read_opc_value(address: str | int) -> any
def read_opc_value(namespace_index: int, address: str | int) -> any
参数:
address: str类型,读取的地址位namespace_index: int类型,命名空间索引(可选)
返回值: 读到的值
异常: 读取失败会抛出RuntimeError异常
read_opc_value_by_subscription
描述: 通过订阅方式读取OPC server中的值
def read_opc_value_by_subscription(namespace_index: int, address: str) -> any
write_opc_value
描述: 向OPC server中的某个地址位写入值
def write_opc_value(address: str | int, value: any) -> bool
def write_opc_value(namespace_index: int, address: str | int, value: any) -> bool
返回值: 成功返回True,失败返回False
S7协议
数据类型读取函数
read_s7_int
描述: S7读取Int
def read_s7_int(plc_type: str, ip: str, block_and_offset: str) -> Optional[int]
参数:
plc_type: str类型,PLC类型,可选值:S1200/S300/S400/S1500/S200Smart/S200ip: str类型,PLC IPblock_and_offset: str类型,读取的地址
支持的地址类型:
| 地址名称 | 地址代号 | 示例 |
|---|---|---|
| 中间寄存器 | M | M100,M200 |
| 输入寄存器 | I | I100,I200 |
| 输出寄存器 | Q | Q100,Q200 |
| DB块寄存器 | DB | DB1.100,DB1.200.7 |
| V寄存器 | V | V100,V200 |
read_s7_dint
描述: S7读取DInt
def read_s7_dint(plc_type: str, ip: str, block_and_offset: str) -> Optional[int]
read_s7_word
描述: S7读取Word
def read_s7_word(plc_type: str, ip: str, block_and_offset: str) -> Optional[int]
read_s7_dword
描述: S7读取DWord
def read_s7_dword(plc_type: str, ip: str, block_and_offset: str) -> Optional[int]
read_s7_string
描述: S7读取String
def read_s7_string(plc_type: str, ip: str, block_and_offset: str) -> Optional[str]
read_s7_boolean
描述: S7读取Bool
def read_s7_boolean(plc_type: str, ip: str, block_and_offset: str) -> Optional[bool]
数据类型写入函数
write_s7_int
描述: S7写入Int
def write_s7_int(plc_type: str, ip: str, block_and_offset: str, value: int) -> bool
write_s7_dint
描述: S7写入DInt
def write_s7_dint(plc_type: str, ip: str, block_and_offset: str, value: int) -> bool
write_s7_word
描述: S7写入Word
def write_s7_word(plc_type: str, ip: str, block_and_offset: str, value: int) -> bool
write_s7_dword
描述: S7写入DWord
def write_s7_dword(plc_type: str, ip: str, block_and_offset: str, value: int) -> bool
write_s7_string
描述: S7写入String
def write_s7_string(plc_type: str, ip: str, block_and_offset: str, value: str) -> bool
write_s7_boolean
描述: S7写入Bool
def write_s7_boolean(plc_type: str, ip: str, block_and_offset: str, value: bool) -> bool
通用读写函数
read_s7
描述: S7通用读取
def read_s7(plc_type: str, ip: str, slot: int, rack: int, data_type: str, block_and_offset: str) -> Any
参数:
slot: CPU模块的插槽号rack: 中央机架data_type: 在Int、DInt、Word、DWord、String、Bool中可选
write_s7
描述: S7通用写入
def write_s7(plc_type: str, ip: str, slot: int, rack: int, data_type: str, block_and_offset: str, value: Any) -> bool
Websocket协议
send_msg_to_wsc_by_client_ip
描述: 根据IP向客户端发送消息
def send_msg_to_wsc_by_client_ip(msg: str, ip: str) -> None
参数:
msg: 字符串类型,表示要发送的消息ip: 字符串类型,表示接收消息的客户端IP
send_msg_to_wsc_by_client_name
描述: 根据客户端名称向客户端发送消息
def send_msg_to_wsc_by_client_name(msg: str, client_name: str) -> None
参数:
msg: 字符串类型,表示要发送的消息client_name: 字符串类型,表示接收消息的客户端名称
get_websocket_client_ip
描述: 获取所有WebSocket客户端的IP
def get_websocket_client_ip() -> list[str]
返回值: 返回一个字符串列表,包含所有客户端的IP
get_websocket_client_name
描述: 获取所有WebSocket客户端名称
def get_websocket_client_name() -> list[str]
返回值: 返回一个字符串列表,包含所有客户端名称
任务管理
check_task_param
描述: 校验输入参数中的task_id和task_label字段对应的数据是否在数据库中存在
async def check_task_param(param: str) -> bool
参数: param是JSON字符串,包含task_id和task_label字段
返回值: True:任务数据存在,False:不存在
create_wind_task
描述: 异步创建并运行一个任务
async def create_wind_task(param: str) -> str
参数: param是JSON字符串,包含:
- task_id: 任务类型id
- task_label: 任务名称
- task_record_id: 任务实例id
- input_params: 任务参数(JSON字符串)
返回值: JSON字符串,包含code(200表示成功,-1表示失败)和msg字段
get_task_record_by_id
描述: 根据任务实例ID查询任务实例
async def get_task_record_by_id(task_record_id: str) -> str
参数: task_record_id,任务实例的唯一ID
返回值: 存在返回JSON格式字符串,不存在返回"null"
sync_create_wind_task
描述: 创建并运行一个任务(阻塞方法)
async def sync_create_wind_task(task_param: str) -> str
返回值: "9013":任务正常结束,"9014":任务异常结束
terminate_tasks
描述: 终止任务
async def terminate_tasks(param: str) -> None
参数: param是JSON数组字符串,包含要终止的任务记录ID列表
mark_complete
描述: 任务封口
async def mark_complete(task_id: str, auto_get_token: bool = True) -> Dict[str, Any]
返回值: 字典类型,包含封口结果信息
库位管理
is_point_exist
描述: 检查站点是否存在
async def is_point_exist(point_name: str) -> bool
check_site_existed_by_site_id
描述: 检查库位是否存在
async def check_site_existed_by_site_id(site_id: str) -> bool
find_sites_by_condition
描述: 根据条件获取库位
async def find_sites_by_condition(conditions: str, sort: str) -> str
参数:
conditions: 查询条件JSON字符串,支持的字段:- area: 区域名
- content: 货物
- filled: 是否有货
- group_names: 库区名
- locked: 是否已锁定
- site_ids: 库位id列表
- tags: 库位标签
sort: 排序方式,"ASC"顺序排序,"DESC"倒序排序
返回值: 库位列表JSON字符串
使用示例:
# 查询库位id为E1-1且有货的库位
condition = {
"site_ids": ["E1-1"],
"filled": True
}
site = await find_sites_by_condition(json.dumps(condition), "ASC")
update_sites_by_condition
描述: 根据条件更新库位
async def update_sites_by_condition(conditions: str, values: str) -> int
参数:
conditions: 更新条件JSON字符串values: 修改的字段JSON字符串
返回值: 更新成功的数据库记录行数
使用示例:
# 对库位E1-1进行加锁
condition = {"site_ids": ["E1-1"], "locked": False}
value = {"locked": True, "locked_by": "task123"}
result = await update_sites_by_condition(json.dumps(condition), json.dumps(value))
find_available_sites_by_condition
描述: 根据条件获取有效库位(未禁用且同步成功的库位)
async def find_available_sites_by_condition(conditions: str, sort: str) -> str
update_site_ext_field_by_id_and_ext_field_name
描述: 根据条件更新库位扩展字段值
async def update_site_ext_field_by_id_and_ext_field_name(conditions: str) -> None
机器人管理
is_robot_exist
描述: 校验机器人是否存在
async def is_robot_exist(robot_name: str) -> bool
get_lite_robots_status
描述: 获取机器人基本信息
async def get_lite_robots_status() -> str
返回值: 包含所有机器人详细信息的JSON数组字符串
get_lite_robots_status_by_id
描述: 根据ID获取机器人基本信息
async def get_lite_robots_status_by_id(agv_id: str) -> str
参数: agv_id,机器人ID或名称
返回值: 机器人信息JSON字符串,未找到返回空对象{}
get_vehicle_station
描述: 根据ID获取机器人位置
async def get_vehicle_station(agv_id: str) -> str
返回值: 包含last_station和station字段的JSON字符串
query_charge_param
描述: 查询所有机器人充电阈值
async def query_charge_param() -> Optional[str]
返回值: 包含机器人充电参数的JSON数组字符串
modify_charge_param
描述: 修改机器人充电阈值
async def modify_charge_param(param: str) -> bool
参数: param,包含机器人ID和充电参数的JSON字符串
支持的参数字段:
- exchange_power: 换货电量阈值
- task_power: 任务电量阈值
- charge_power: 充电电量阈值
- min_power: 最小电量
- max_power: 最大电量
缓存数据管理
get_cache_param
描述: 获取缓存数据
async def get_cache_param(key: str) -> str | None
参数: key,缓存键
返回值: JSON格式字符串,不存在返回None
put_cache_param
描述: 缓存数据
async def put_cache_param(key: str, value: str) -> None
参数:
key: 缓存键value: 缓存值,如果要缓存对象需要先序列化为JSON字符串
clear_cache_param
描述: 删除缓存数据
async def clear_cache_param(key: str) -> None
get_all_cache_params
描述: 获取所有的缓存数据
async def get_all_cache_params() -> str
返回值: JSON格式字符串,不存在返回空字符串
系统工具
create_uuid
描述: 创建唯一的ID编号
def create_uuid() -> str
返回值: 表示一个唯一ID编号的字符串
json_to_xml
描述: 将JSON字符串转化为XML字符串
def json_to_xml(param: str) -> str
md5_encode
描述: MD5字符串加密
def md5_encode(param: str) -> str
返回值: 32位小写字符串
base64_encode
描述: Base64字符串重编码
def base64_encode(param: str) -> str
返回值: UrlBase64重编码后的字符串
trans_to_page_obj
描述: 转换成分页对象
def trans_to_page_obj(total_count: int, current_page: int, page_size: int, total_page: int, page_list: str) -> str
参数:
total_count: 分页数据的总条数current_page: 当前是第几页page_size: 每页显示的数量total_page: 总页数page_list: 当前页的数据
get_settings_config_value
描述: 获取settings.py的配置
def get_settings_config_value(key: str) -> str
参数: key,settings.py中所要获取配置的字段名
返回值: JSON字符串
数据库操作
jdbc_execute_sql
描述: 执行SQL语句
def jdbc_execute_sql(sql: str) -> bool
参数: sql,创建表的SQL语句
返回值: True:SQL命令执行成功,False:执行失败
jdbc_query
描述: 使用SQL查询数据
def jdbc_query(sql: str) -> str
参数: sql,SQL查询语句
返回值: JSON格式的查询结果字符串
jdbc_insert_or_update
描述: 更新数据
def jdbc_insert_or_update(sql: str, *sql_params: Any) -> int
参数:
sql: SQL语句字符串sql_params: 可动态扩展的参数
返回值: 0表示更新失败,大于0的数字表示更新的记录数目
使用示例:
sql = "update employee set age = 25 where name = ? and id = ?"
jdbc_insert_or_update(sql, "bob", 123)
jdbc_query_count
描述: 查询数据数量
def jdbc_query_count(sql: str) -> Optional[int]
参数: sql,SQL语句字符串(只能使用select count语句)
返回值: None表示查询失败,其他数字表示查询到的记录数目
文件操作
read_file_to_string
描述: 根据文件名读取脚本目录下的文件内容
def read_file_to_string(file_name: str) -> Optional[str]
参数: file_name,目标文件的名称(带后缀,例如a.txt)
返回值: 成功返回文件内容字符串,失败返回None
日志相关
get_logger
描述: 获取日志打印对象
def get_logger() -> Logger
返回值: 日志对象,包含info、debug、warn、error方法
使用示例:
get_logger().info("hello world")
get_logger().error("there is an error.", Exception("错误详情"))
script_log
描述: 在脚本页面下打印日志
def script_log(level: str, function_name: str, content: any) -> None
参数:
level: 日志等级,可选"INFO"、"ERROR"function_name: 函数签名,用于快速定位信息content: 日志信息
注册方法
注意:以下方法需要在boot()函数中调用
define_scheduled_functions
描述: 定时执行脚本方法
def define_scheduled_functions(is_parallel: bool, delay: int, period: int, function_name: str, args: list) -> None
参数:
is_parallel: 是否并行delay: 启动延迟执行的时间,单位毫秒period: 脚本方法的周期,单位毫秒function_name: 周期执行的方法名args: 周期执行的方法的参数
register_button
描述: 向RDS注册管理后台可见的按钮
def register_button(label: str, remark: str, script_function: str, level: str, type: int = 0) -> None
参数:
label: 按钮名remark: 按钮备注script_function: 点击按钮触发的方法名(注意该方法不能带参数)level: 按钮等级,用颜色区分(blue:蓝色,yellow:黄色,red:红色)type: 0:普通按钮,1:导出按钮
register_handler
描述: 在脚本中注册接口方法
def register_handler(method: str, path: str, callback: str, auth: bool) -> None
参数:
method: HTTP请求方式,支持POST、GETpath: 接口的URL路径callback: 处理接口的脚本方法名auth: 调用此接口是否需要登录
多线程及时间
sleep
描述: 休眠指定时间
async def sleep(ms: int) -> None
参数: ms,休眠的时间,单位毫秒
使用示例:
# 异步休眠1秒
await sleep(1000)
current_time_millis
描述: 获取当前系统时间戳
def current_time_millis() -> int
返回值: 当前系统的毫秒时间戳
now_date
描述: 获取当前时间
def now_date() -> str
返回值: 按照yyyy-MM-dd HH:mm:ss格式的字符串
time_millis_format
描述: 将时间戳转为日期格式
def time_millis_format(time: int) -> str
参数: time,毫秒时间戳
返回值: 按照yyyy-MM-dd HH:mm:ss格式的字符串
手持端相关
通知功能
notice_operator
描述: 通知手持端(工位/岗位)
function noticeOperator(workTypes: string, workStations: string, content: string, needConfirm: boolean, keepTime: number): void
参数:
workTypes: 要通知的岗位,多岗位间逗号分隔workStations: 要通知的工位,多工位间逗号分隔content: 通知的内容needConfirm: 是否需要手持端用户点击确认后通知弹框才会消失keepTime: 通知弹框在手持端的保持时间,单位秒
notice_operator_by_user
描述: 通知手持端用户
function noticeOperatorByUser(userNames: string, content: string, needConfirm: boolean, keepTime: number): void
界面控制
set_pad_label_by_menu_id
描述: 通过menuId修改手持端按钮的label
function setPadLabelByMenuId(menuId: String, label: String): void
set_pad_background_by_menu_id
描述: 通过menuId修改手持端按钮的背景色
function setPadBackgroundByMenuId(menuId: String, background: String): void
set_pad_disable_by_menu_id
描述: 通过menuId设置手持端按钮是否禁用
function setPadDisableByMenuId(menuId: String, disable: boolean): void
总结
本文档涵盖了VWED在线脚本系统中所有可用的内置函数,按功能模块进行了详细分类。开发者可以根据实际需求选择合适的函数来实现业务逻辑。
主要特性:
- 协议通信:支持多种工业协议(Modbus、S7、OPC-UA、MQTT等)
- 任务管理:完整的任务生命周期管理
- 库位管理:灵活的仓储库位操作
- 机器人管理:AMR机器人状态监控和控制
- 系统工具:丰富的系统级工具函数
- 数据库操作:直接的SQL查询和操作能力
使用建议:
- 异步函数需要使用await关键字调用
- 注册相关的函数必须在boot()函数中调用
- 错误处理要根据各函数的异常说明进行相应处理
- 建议在使用前先了解相关协议和系统的基础知识