【最佳实践-视觉智能】视觉智能之ArkClaw控制IPC - 文章 - 开发者社区

ArkClaw在IPC领域，作为设备的控制枢纽，可以把设备控制能力封装为ArkClaw的Skill。
用户提问时，ArkClaw自主决策调用对应Skill执行任务。其强大的Skill编排能力，可以胜任复杂的任务，比如“帮忙找到家里的小孩，看他在做什么”

方案

picture.image

业务流程

picture.image

SKILL建设思路

技能名称：ipc-capture

作用：抓图

技能说明

---
name: ipc-capture
description: 使用内置 ipc_capture.py 脚本抓IPC摄像头的监控图片，支持将图片保存到本地目录，返回图片URL以及图片分析结果。
---

# IPC CAPTURE

## 适用场景
- 远程监控特定区域的实时状态
- 定期抓取监控图片进行存档
- 事件或指令触发抓取现场图片
- 需要获取摄像头实时画面进行分析

## 使用步骤

1. 运行脚本 `python scripts/ipc_capture.py`。运行之前cd到对应的目录。
2. 查看输出结果，获取图片URL、本地保存路径、图片分析结果。

## 认证与凭据来源

- 读取 `ACCESS_TOKEN` 环境变量。

## 输出格式

- 输出脚本生成的图片URL、本地保存路径、图片分析结果。
- 若调用失败，将打印错误信息。

## 示例

```bash
python scripts/ipc_capture.py
```

scripts

# 控制IPC摄像头抓拍图片，参考接口文档https://open.ys7.com/help/687
import requests
import json
import os
from volcenginesdkarkruntime import Ark

class IpcCapture:
    def __init__(self, access_token=None):
        """初始化IPC抓图类"""
        self.base_url = "https://open.ys7.com"
        # 优先使用传入的access_token，否则从环境变量获取
        self.access_token = access_token or os.getenv('YS7_ACCESS_TOKEN')
        if not self.access_token:
            raise ValueError("未提供access_token且环境变量YS7_ACCESS_TOKEN未设置")
        self.capture_endpoint = "/api/lapp/device/capture"
    
    def download_image(self, image_url, save_path):
        """
        下载图片并保存到本地
        :param image_url: 图片URL
        :param save_path: 保存路径
        :return: 是否下载成功
        """
        try:
            print(f"正在下载图片: {image_url}")
            response = requests.get(image_url)
            response.raise_for_status()
            
            with open(save_path, 'wb') as f:
                f.write(response.content)
            
            print(f"图片已成功保存到: {save_path}")
            return True
        except Exception as e:
            print(f"下载图片失败: {str(e)}")
            return False
    
    def analyze_image(self, image_url):
        """
        调用火山方舟Responses API分析图片内容
        :param image_url: 图片URL
        :return: 分析结果
        """
        # 获取火山方舟API Key
        ark_api_key = os.getenv('ARK_API_KEY')
        if not ark_api_key:
            print("未设置环境变量ARK_API_KEY，无法分析图片")
            return None
        
        try:
            print("正在分析图片内容...")
            
            # 创建Ark客户端
            client = Ark(
                base_url='https://ark.cn-beijing.volces.com/api/v3',
                api_key=ark_api_key,
            )
            
            # 调用responses.create方法
            response = client.responses.create(
                model="doubao-seed-2-0-mini-260215",
                input=[
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "input_image",
                                "image_url": image_url
                            },
                            {
                                "type": "input_text",
                                "text": "请分析这张图片的内容，输出结果控制在100字以内"
                            },
                        ]
                        
                    }
                ],
                thinking={'type':'disabled'}
            )
            
            # 打印响应
            print(f"图片分析接口响应: {response}")
            
            # 提取分析结果
            analysis_result = response.output[0].content[0].text
            print(f"图片分析结果: {analysis_result}")
            return analysis_result
        
        except Exception as e:
            print(f"请求失败: {str(e)}")
            return None
    
    def capture_image(self, device_serial, channel_no=1, quality=1, save_local=True, analyze=True):
        """
        调用设备抓拍图片接口
        :param device_serial: 设备序列号，字母需为大写
        :param channel_no: 通道号，IPC设备填写1
        :param quality: 视频清晰度，0-流畅，1-高清(720P)，2-4CIF，3-1080P，4-400w（注：此参数不生效）
        :param save_local: 是否保存到本地
        :param analyze: 是否分析图片内容
        :return: 响应结果，包含图片URL、本地保存路径和分析结果
        """
        url = f"{self.base_url}{self.capture_endpoint}"
        
        # 构建请求参数
        data = {
            "accessToken": self.access_token,
            "deviceSerial": device_serial.upper(),  # 确保设备序列号为大写
            "channelNo": channel_no,
            "quality": quality
        }
        
        # 构建请求头
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        
        try:
            print(f"正在调用抓图接口，设备序列号: {device_serial}, 通道号: {channel_no}")
            response = requests.post(url, data=data, headers=headers)
            response.raise_for_status()  # 检查HTTP响应状态
            
            # 解析响应
            result = response.json()
            print(f"抓图接口响应: {json.dumps(result, ensure_ascii=False, indent=2)}")
            
            if result.get("code") == "200":
                print("抓图成功！")
                pic_url = result.get("data", {}).get("picUrl")
                
                # 保存到本地
                local_path = None
                if save_local and pic_url:
                    # 生成保存文件名，使用更易读的时间戳格式
                    import time
                    timestamp = time.strftime("%Y年%m月%d日%H点%M分%S秒")
                    filename = f"ipc_capture_{device_serial}_{timestamp}.jpg"
                    local_path = os.path.join(os.getcwd(), filename)
                    
                    # 下载并保存图片
                    self.download_image(pic_url, local_path)
                
                # 分析图片内容
                analysis_result = None
                if analyze and pic_url:
                    analysis_result = self.analyze_image(pic_url)
                
                # 返回结果
                return {
                    "code": "200",
                    "picUrl": pic_url,
                    "localPath": local_path,
                    "analysisResult": analysis_result
                }
            else:
                print(f"抓图失败: {result.get('msg', '未知错误')}")
                return result
        
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {str(e)}")
            return {"code": "500", "msg": f"请求失败: {str(e)}"}

if __name__ == "__main__":
    # 使用示例
    # 请替换为实际的设备序列号
    # access_token将从环境变量YS7_ACCESS_TOKEN获取
    device_serial = "BD7000602"
    
    try:
        ipc_capture = IpcCapture()
        result = ipc_capture.capture_image(device_serial)
        
        # 处理返回结果
        if result.get("code") == "200":
            print("抓图操作已成功完成")
            print(f"图片URL: {result.get('picUrl')}")
            if result.get('localPath'):
                print(f"图片已保存到: {result.get('localPath')}")
            if result.get('analysisResult'):
                print(f"图片分析结果: {result.get('analysisResult')}")
        else:
            print("抓图操作失败")
            if result.get('msg'):
                print(f"失败原因: {result.get('msg')}")
    except ValueError as e:
        print(f"错误: {str(e)}")
        print("请设置环境变量YS7_ACCESS_TOKEN，例如:")
        print("export YS7_ACCESS_TOKEN=your_access_token_here")

技能名称：云台控制

作用：调整摄像头方向

技能说明：

---
name: ipc-ptz-control
description: 使用内置 ipc_ptz_control.py 脚本来调整摄像头方向，需要输入调整方向和调整速度。
---

# IPC PTZ CONTROL

## 适用场景
- 远程监控特定区域的实时状态
- 定期调整摄像头方向以监控不同区域
- 事件或指令触发调整摄像头方向
- 需要手动调整摄像头方向以监控不同区域

## 使用步骤

1. 运行脚本 `python scripts/ipc_ptz_control.py`。运行之前cd到对应的目录。
2. 查看输出结果，获取图片URL和本地保存路径。

## 输入参数

- `direction`: 调整方向，类型为int，可选值为 0-上，1-下，2-右，3-左。
- `speed`: 调整速度，类型为int，可选值为 0-慢，1-适中，2-快。

## 认证与凭据来源

- 读取 `ACCESS_TOKEN` 环境变量。

## 输出格式

- 输出调整结果，成功返回调整结果，失败返回错误信息。

## 示例

```bash
python scripts/ipc_ptz_control.py --direction 0 --speed 1
```

scripts：

# 控制IPC摄像头方向，参考接口文档https://open.ys7.com/help/679
import requests
import json
import os

class IpcPtzControl:
    def __init__(self, access_token=None):
        """初始化IPC PTZ控制类"""
        self.base_url = "https://open.ys7.com"
        # 优先使用传入的access_token，否则从环境变量获取
        self.access_token = access_token or os.getenv('YS7_ACCESS_TOKEN')
        if not self.access_token:
            raise ValueError("未提供access_token且环境变量YS7_ACCESS_TOKEN未设置")
        self.ptz_start_endpoint = "/api/lapp/device/ptz/start"
        self.ptz_stop_endpoint = "/api/lapp/device/ptz/stop"
    
    def start_ptz_control(self, device_serial, channel_no=1, direction=1, speed=2):
        """
        开始云台控制
        :param device_serial: 设备序列号，字母需为大写
        :param channel_no: 通道号，IPC设备填写1
        :param direction: 控制方向，0-上，1-下，2-右，3-左
        :param speed: 控制速度，1-慢，2-中，3-快
        :return: 响应结果
        """
        url = f"{self.base_url}{self.ptz_start_endpoint}"
        
        # 构建请求参数
        data = {
            "accessToken": self.access_token,
            "deviceSerial": device_serial.upper(),  # 确保设备序列号为大写
            "channelNo": channel_no,
            "direction": direction,
            "speed": speed
        }
        
        # 构建请求头
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        
        try:
            print(f"正在调用开始云台控制接口，设备序列号: {device_serial}, 方向: {direction}, 速度: {speed}")
            response = requests.post(url, data=data, headers=headers)
            response.raise_for_status()  # 检查HTTP响应状态
            
            # 解析响应
            result = response.json()
            print(f"开始云台控制接口响应: {json.dumps(result, ensure_ascii=False, indent=2)}")
            
            if result.get("code") == "200":
                print("开始云台控制成功！")
                return result
            else:
                print(f"开始云台控制失败: {result.get('msg', '未知错误')}")
                return result
        
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {str(e)}")
            return {"code": "500", "msg": f"请求失败: {str(e)}"}
    
    def stop_ptz_control(self, device_serial, channel_no=1):
        """
        停止云台控制
        :param device_serial: 设备序列号，字母需为大写
        :param channel_no: 通道号，IPC设备填写1
        :return: 响应结果
        """
        url = f"{self.base_url}{self.ptz_stop_endpoint}"
        
        # 构建请求参数
        data = {
            "accessToken": self.access_token,
            "deviceSerial": device_serial.upper(),  # 确保设备序列号为大写
            "channelNo": channel_no
        }
        
        # 构建请求头
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        
        try:
            print(f"正在调用停止云台控制接口，设备序列号: {device_serial}")
            response = requests.post(url, data=data, headers=headers)
            response.raise_for_status()  # 检查HTTP响应状态
            
            # 解析响应
            result = response.json()
            print(f"停止云台控制接口响应: {json.dumps(result, ensure_ascii=False, indent=2)}")
            
            if result.get("code") == "200":
                print("停止云台控制成功！")
                return result
            else:
                print(f"停止云台控制失败: {result.get('msg', '未知错误')}")
                return result
        
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {str(e)}")
            return {"code": "500", "msg": f"请求失败: {str(e)}"}
    
    def control_camera(self, device_serial, direction, duration=1, speed=1):
        """
        控制摄像头方向，自动开始和停止
        :param device_serial: 设备序列号，字母需为大写
        :param direction: 控制方向，0-上，1-下，2-右，3-左
        :param duration: 控制持续时间（秒）
        :param speed: 控制速度，1-慢，2-中，3-快
        :return: 响应结果
        """
        # 开始云台控制
        start_result = self.start_ptz_control(device_serial, direction=direction, speed=speed)
        
        if start_result.get("code") == "200":
            # 等待指定时间
            import time
            print(f"控制摄像头{duration}秒...")
            time.sleep(duration)
            
            # 停止云台控制
            stop_result = self.stop_ptz_control(device_serial)
            return stop_result
        else:
            return start_result

if __name__ == "__main__":
    # 使用示例
    # 请替换为实际的设备序列号
    # access_token将从环境变量YS7_ACCESS_TOKEN获取
    device_serial = "BD7000602"
    
    try:
        ptz_control = IpcPtzControl()
        
        # 控制摄像头移动1秒
        print("\n=== 测试摄像头转动 ===")
        result = ptz_control.control_camera(device_serial, direction=10, duration=1, speed=2)
        
        # 处理返回结果
        if result.get("code") == "200":
            print("摄像头控制操作已成功完成")
        else:
            print("摄像头控制操作失败")
            if result.get('msg'):
                print(f"失败原因: {result.get('msg')}")
    except ValueError as e:
        print(f"错误: {str(e)}")
        print("请设置环境变量YS7_ACCESS_TOKEN，例如:")
        print("export YS7_ACCESS_TOKEN=your_access_token_here")

技能名称：tos-access

作用：对象存储的文件上传和下载

技能说明：

---
name: tos-access
description: 使用内置 main.py 脚本来操作TOS，实现文件上传和下载功能，上传需要输入文件路径，下载需要输入文件的tos url。
---

# TOS文件操作技能

## 功能描述

TOS文件操作技能是一个基于火山引擎对象存储服务（TOS）的工具，提供了文件上传和下载的功能。通过命令行界面，用户可以方便地将本地文件上传到TOS，或从TOS下载文件到本地。

## 核心功能

- **文件上传** ：将本地文件上传到TOS指定存储桶
- **文件下载** ：从TOS指定存储桶下载文件到本地
- **URL解析** ：自动解析TOS URL格式，提取存储桶和对象信息
- **目录自动创建** ：下载时自动创建不存在的目录结构
- **环境变量配置** ：支持从环境变量获取认证信息
- **详细的错误处理** ：捕获并显示各种可能的错误信息

## 技术实现

- 使用火山引擎TOS Python SDK进行文件操作
- 采用模块化设计，将上传和下载功能分离
- 使用argparse库实现命令行参数解析
- 支持子命令模式，提供download和upload两个操作
- 提供清晰的返回值，便于调用方处理结果

## 安装和配置

### 环境要求

- Python 3.7+
- 火山引擎TOS SDK：`tos`

### 安装依赖

```bash
pip install tos
```

### 配置认证信息

需要设置以下环境变量：

```bash
export VOLCENGINE_ACCESS_KEY=your_access_key
export VOLCENGINE_SECRET_KEY=your_secret_key
```

### 配置存储桶信息

默认配置：
- Endpoint: tos-cn-beijing.volces.com
- Region: cn-beijing
- Bucket Name: xujianhua-utils
- 上传路径前缀: visual_intelligence/

## 使用方法

### 命令行格式

```bash
python main.py <action> [options]
```

### 操作类型

- `download`：从TOS下载文件
- `upload`：上传文件到TOS

### 下载操作

**参数** ：
- `--tos-url` 或 `-u`：TOS URL，格式为 `tos://bucket_name/object_key`（必填）

**示例** ：
```bash
python main.py download --tos-url "tos://xujianhua-utils/visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg"
```

### 上传操作

**参数** ：
- `--file` 或 `-f`：本地文件路径（必填）

**示例** ：
```bash
python main.py upload --file "/root/vlm/SKILLS/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg"
```

## 工作流程

### 下载流程
1. 解析命令行参数，获取TOS URL
2. 解析TOS URL，提取存储桶名称和对象键
3. 构建本地保存路径
4. 确保保存目录存在
5. 创建TOS客户端并执行下载操作
6. 返回本地文件保存路径

### 上传流程
1. 解析命令行参数，获取本地文件路径
2. 构建TOS对象键（使用文件名）
3. 创建TOS客户端并执行上传操作
4. 返回完整的TOS URL

## 注意事项

- 确保环境变量中设置了正确的访问密钥
- 上传操作会将文件保存到 `visual_intelligence/` 前缀下
- 下载操作会将文件保存到 `../` 目录下，保持与对象键相同的目录结构
- TOS URL格式必须为 `tos://bucket_name/object_key`
- 确保网络连接稳定，特别是处理大文件时

## 故障排除

- **认证错误** ：确保环境变量中设置了正确的 `VOLCENGINE_ACCESS_KEY` 和 `VOLCENGINE_SECRET_KEY`
- **网络错误** ：检查网络连接，确保能够访问TOS服务
- **权限错误** ：确保使用的访问密钥具有操作对应存储桶的权限
- **路径错误** ：确保本地文件路径存在且可访问
- **格式错误** ：确保TOS URL格式正确

## 示例输出

### 下载操作

```
执行下载操作: tos://xujianhua-utils/visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg
正在下载文件到: ../visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg
文件已成功下载到: ../visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg
下载成功，保存路径: ../visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg
```

### 上传操作

```
执行上传操作: /root/vlm/SKILLS/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg
文件 /root/vlm/SKILLS/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg 已成功上传到 xujianhua-utils/visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg
上传成功，tos_url: tos://xujianhua-utils/visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg
```

scripts

import os
import tos


def parse_tos_url(tos_url):
    """
    解析TOS URL，提取bucket_name和object_key
    :param tos_url: TOS URL，格式为 "tos://bucket_name/object_key"
    :return: (bucket_name, object_key)
    """
    # 移除 "tos://" 前缀
    if tos_url.startswith("tos://"):
        tos_url = tos_url[6:]
    
    # 分割bucket_name和object_key
    parts = tos_url.split("/", 1)
    if len(parts) != 2:
        raise ValueError("Invalid TOS URL format. Expected format: tos://bucket_name/object_key")
    
    bucket_name = parts[0]
    object_key = parts[1]
    
    return bucket_name, object_key


def download_file_from_tos(tos_url):
    """
    从TOS下载文件到本地
    :param tos_url: TOS URL，格式为 "tos://bucket_name/object_key"
    :return: 本地文件保存路径
    """
    # 从环境变量获取 AK 和 SK 信息。
    ak = os.getenv('VOLCENGINE_ACCESS_KEY')
    sk = os.getenv('VOLCENGINE_SECRET_KEY')
    # your endpoint 和 your region 填写Bucket 所在区域对应的Endpoint。# 以华北2(北京)为例，your endpoint 填写 tos-cn-beijing.volces.com，your region 填写 cn-beijing。
    endpoint = "tos-cn-beijing.volces.com"
    region = "cn-beijing"
    
    try:
        # 解析TOS URL
        bucket_name, object_key = parse_tos_url(tos_url)
        
        # 本地文件保存路径
        file_name = f"../{object_key}"
        
        # 确保保存目录存在
        os.makedirs(os.path.dirname(file_name), exist_ok=True)
        
        print(f"正在下载文件到: {file_name}")
        
        # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
        client = tos.TosClientV2(ak, sk, endpoint, region)
        # 若 file_name 为目录则将对象下载到此目录下, 文件名为对象名
        client.get_object_to_file(bucket_name, object_key, file_name)
        
        print(f"文件已成功下载到: {file_name}")
        return file_name
        
    except tos.exceptions.TosClientError as e:
        # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
        print('fail with client error, message:{}, cause: {}'.format(e.message, e.cause))
        return None
    except tos.exceptions.TosServerError as e:
        # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
        print('fail with server error, code: {}'.format(e.code))
        # request id 可定位具体问题，强烈建议日志中保存
        print('error with request id: {}'.format(e.request_id))
        print('error with message: {}'.format(e.message))
        print('error with http code: {}'.format(e.status_code))
        print('error with ec: {}'.format(e.ec))
        print('error with request url: {}'.format(e.request_url))
        return None
    except Exception as e:
        print('fail with unknown error: {}'.format(e))
        return None

import tos


def upload_file_to_tos(file_name):
    """
    上传文件到TOS
    :param file_name: 本地文件路径
    :return: object_key
    """
    # 从环境变量获取 AK 和 SK 信息。
    ak = os.getenv('VOLCENGINE_ACCESS_KEY')
    sk = os.getenv('VOLCENGINE_SECRET_KEY')
    # your endpoint 和 your region 填写Bucket 所在区域对应的Endpoint。# 以华北2(北京)为例，your endpoint 填写 tos-cn-beijing.volces.com，your region 填写 cn-beijing。
    endpoint = "tos-cn-beijing.volces.com"
    region = "cn-beijing"
    bucket_name = "xujianhua-utils"
    
    # 对象名称，例如 example_dir 下的 example_object.txt 文件，则填写为 example_dir/example_object.txt
    object_key = "visual_intelligence/" + os.path.basename(file_name)
    
    try:
        # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
        client = tos.TosClientV2(ak, sk, endpoint, region)
        # 将本地文件上传到目标桶中
        # file_name为本地文件的完整路径。
        client.put_object_from_file(bucket_name, object_key, file_name)
        print(f"文件 {file_name} 已成功上传到 {bucket_name}/{object_key}")
        return f"tos://{bucket_name}/{object_key}"
        
    except tos.exceptions.TosClientError as e:
        # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
        print('fail with client error, message:{}, cause: {}'.format(e.message, e.cause))
        return None
    except tos.exceptions.TosServerError as e:
        # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
        print('fail with server error, code: {}'.format(e.code))
        # request id 可定位具体问题，强烈建议日志中保存
        print('error with request id: {}'.format(e.request_id))
        print('error with message: {}'.format(e.message))
        print('error with http code: {}'.format(e.status_code))
        print('error with ec: {}'.format(e.ec))
        print('error with request url: {}'.format(e.request_url))
        return None
    except Exception as e:
        print('fail with unknown error: {}'.format(e))
        return None

# TOS操作入口脚本

import argparse
import sys
import os

# 添加当前目录到Python路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from download_file import download_file_from_tos
from upload_file import upload_file_to_tos


def parse_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(description='TOS文件操作工具')
    
    # 子命令解析器
    subparsers = parser.add_subparsers(dest='action', help='操作类型')
    
    # 下载子命令
    download_parser = subparsers.add_parser('download', help='从TOS下载文件')
    download_parser.add_argument('--tos-url', '-u', type=str, required=True, help='TOS URL，格式为 tos://bucket_name/object_key')
    
    # 上传子命令
    upload_parser = subparsers.add_parser('upload', help='上传文件到TOS')
    upload_parser.add_argument('--file', '-f', type=str, required=True, help='本地文件路径')
    
    return parser.parse_args()


def main():
    """主函数"""
    try:
        # 解析命令行参数
        args = parse_args()
        
        if args.action == 'download':
            # 执行下载操作
            print(f"执行下载操作: {args.tos_url}")
            local_path = download_file_from_tos(args.tos_url)
            if local_path:
                print(f"下载成功，保存路径: {local_path}")
            else:
                print("下载失败")
                sys.exit(1)
        
        elif args.action == 'upload':
            # 执行上传操作
            print(f"执行上传操作: {args.file}")
            tos_url = upload_file_to_tos(args.file)
            if tos_url:
                print(f"上传成功，tos_url: {tos_url}")
            else:
                print("上传失败")
                sys.exit(1)
        
        else:
            print("请指定操作类型: download 或 upload")
            sys.exit(1)
            
    except Exception as e:
        print(f"发生错误: {str(e)}")
        sys.exit(1)


if __name__ == "__main__":
    main()
    # 使用示例:
    # 下载: python main.py download --tos-url "tos://xujianhua-utils/visual_intelligence/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg"
    # 上传: python main.py upload --file "/root/vlm/SKILLS/ipc_capture_BD7000602_2026年03月15日19点01分13秒.jpg"

DEMO

设备类型	设备序列号	描述
xx型号	xxxxx	一款搭载屏幕、云台和摄像头的IPC设备

效果展示

技能	结果
抓图
云台控制
图像检索