# Katisian Robot

**Repository Path**: null_742_3501/katisian-robot

## Basic Information

- **Project Name**: Katisian Robot
- **Description**: No description available
- **Primary Language**: Python
- **License**: GPL-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-06
- **Last Updated**: 2026-03-25

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# MyYYSTool

一个包含多个实用工具的Python项目，目前包含任务调度器、ADB控制器和图片处理模块。

### 1. 任务调度器 (Task Scheduler)

一个简单易用的Python任务调度器，支持基于优先级和时间的任务调度。

**主要特性：**
- ✅ 基于优先级和时间的任务排序
- ✅ 支持定时重复任务
- ✅ 任务执行超时控制
- ✅ 任务失败重试机制
- ✅ 动态任务管理
- ✅ 调度器控制（启动、停止、暂停、恢复）
- ✅ 线程安全
- ✅ 详细日志和统计信息

**快速开始：**
```python
from datetime import datetime
from task_scheduler import TaskScheduler

scheduler = TaskScheduler()

def my_task(task):
    print(f"执行任务: {task.name}")
    return "完成"

task_id = scheduler.add_task(
    name="我的任务",
    priority=5,
    execution_time=datetime.now(),
    interval=0,
    func=my_task
)

scheduler.start()
```

详细文档请参考：[task_scheduler/README.md](task_scheduler/README.md)

### 2. ADB控制器 (ADB Controller)

通过ADB（Android Debug Bridge）连接Android设备，提供设备控制功能。

**主要特性：**
- ✅ 设备连接管理（USB和网络连接）
- ✅ 应用管理（安装、卸载、查询）
- ✅ 输入控制（点击、按住、滑动、文字输入）
- ✅ 设备截图
- ✅ 设备信息查询

**快速开始：**
```python
from adb_controller import ADBController

controller = ADBController.create()

# 获取设备列表
devices = controller.get_devices()

# 点击屏幕
controller.tap(540, 960)

# 滑动屏幕
controller.swipe(540, 1400, 540, 500, 500)

# 截图
screenshot_path = controller.screenshot()
```

详细文档请参考：[adb_controller/README.md](adb_controller/README.md)

### 3. 图片处理模块 (Image Processor)

处理从ADB控制器获取的设备截图，提供图片查找、文字识别和文字查找功能。

**主要特性：**
- ✅ 图片查找（模板匹配）
- ✅ 文字识别（OCR）
- ✅ 文字查找
- ✅ 截图缓存管理
- ✅ 范围限制操作
- ✅ 精确匹配和模糊匹配

**快速开始：**
```python
from adb_controller import ADBController
from image_processor import ImageProcessor

adb_controller = ADBController.create()
processor = ImageProcessor.create()

# 查找图片
result = processor.find_image("template.png")

# 识别文字
results = processor.recognize_text()

# 查找文字
result = processor.find_text("设置")
```

详细文档请参考：[image_processor/README.md](image_processor/README.md)

## 安装

本项目主要使用Python标准库，部分模块需要额外依赖。

**Python版本要求**：3.8+

**ADB控制器模块额外要求**：需要安装ADB工具（Android SDK Platform Tools）

**图片处理模块额外要求**：需要安装OpenCV和PaddleOCR

```bash
# 克隆项目
git clone <repository-url>

# 进入项目目录
cd MyYYSTool

# 安装图片处理模块依赖
pip install opencv-python paddleocr
```

## 项目结构

```
MyYYSTool/
├── task_scheduler/          # 任务调度器模块
│   ├── __init__.py
│   ├── enums.py
│   ├── models.py
│   ├── validator.py
│   ├── queue.py
│   ├── executor.py
│   ├── scheduler.py
│   └── logger.py
├── adb_controller/          # ADB控制器模块
│   ├── __init__.py
│   ├── enums.py
│   ├── models.py
│   ├── connection.py
│   ├── app_manager.py
│   ├── input_controller.py
│   ├── screenshot.py
│   └── controller.py
├── examples/                # 示例代码
│   ├── basic_usage.py
│   └── advanced_usage.py
├── tests/                   # 测试目录（预留）
├── .codeartsdoer/          # SDD文档目录
├── README.md               # 项目文档
└── requirements.txt        # 依赖文件
```

## 开发指南

### SDD文档

项目使用规范的SDD（Specification-Driven Development）开发流程：

- **spec.md**: 需求规格文档，定义"做什么"
- **design.md**: 设计方案文档，定义"怎么做"
- **tasks.md**: 任务规划文档，定义实施步骤

SDD文档位于 `.codeartsdoer/specs/` 目录下。

### 代码规范

- 使用类型注解
- 完整的文档字符串
- 清晰的变量命名
- 模块化设计
- 完善的异常处理

## 示例代码

### 任务调度器示例

```bash
# 基础使用
python examples/basic_usage.py

# 高级功能
python examples/advanced_usage.py
```

### ADB控制器示例

```bash
# 基础使用
cd adb_controller/examples
python basic_usage.py

# 高级功能
python advanced_usage.py
```

## 注意事项

### 任务调度器
1. 线程安全：调度器内部使用锁保护共享资源
2. 异常处理：任务执行函数应该自行处理异常
3. 资源清理：任务执行函数应该确保正确释放资源

### ADB控制器
1. ADB工具：确保系统已安装ADB工具
2. 设备授权：首次连接设备需要在设备上授权
3. 网络连接：网络连接需要设备和电脑在同一网络
4. 坐标系统：不同设备的屏幕分辨率不同

## 许可证

MIT License

## 贡献

欢迎提交 Issue 和 Pull Request！

## 联系方式

如有问题或建议，请提交 Issue。

## 安装

本项目仅使用Python标准库，无需额外依赖。

**Python版本要求**：3.8+

```bash
# 克隆项目
git clone <repository-url>

# 进入项目目录
cd MyYYSTool

# 无需安装依赖（仅使用标准库）
```

## 快速开始

### 基础示例

```python
from datetime import datetime, timedelta
from task_scheduler import TaskScheduler

# 创建调度器
scheduler = TaskScheduler()

# 定义任务执行函数
def my_task(task):
    print(f"执行任务: {task.name}")
    print(f"任务ID: {task.task_id}")
    print(f"优先级: {task.priority}")
    return "任务完成"

# 添加任务（立即执行）
task_id1 = scheduler.add_task(
    name="高优先级任务",
    priority=10,
    execution_time=datetime.now(),
    interval=0,  # 不重复
    func=my_task
)

# 添加任务（5秒后执行）
task_id2 = scheduler.add_task(
    name="中优先级任务",
    priority=5,
    execution_time=datetime.now() + timedelta(seconds=5),
    interval=0,
    func=my_task
)

# 启动调度器
scheduler.start()

# 等待任务执行
import time
time.sleep(10)

# 停止调度器
scheduler.stop()
```

### 定时重复任务

```python
from datetime import datetime, timedelta
from task_scheduler import TaskScheduler

scheduler = TaskScheduler()

def periodic_task(task):
    print(f"执行周期性任务: {task.name} - {datetime.now()}")

# 添加每10秒执行一次的任务
task_id = scheduler.add_task(
    name="周期性任务",
    priority=5,
    execution_time=datetime.now(),
    interval=10,  # 每10秒执行一次
    func=periodic_task
)

scheduler.start()

# 运行一段时间
time.sleep(30)

scheduler.stop()
```

### 任务重试机制

```python
from datetime import datetime
from task_scheduler import TaskScheduler

scheduler = TaskScheduler()

def unreliable_task(task):
    import random
    if random.random() < 0.7:  # 70%概率失败
        raise Exception("任务执行失败")
    print("任务执行成功！")
    return "成功"

# 添加带重试的任务
task_id = scheduler.add_task(
    name="可能失败的任务",
    priority=8,
    execution_time=datetime.now(),
    interval=0,
    func=unreliable_task,
    retry_count=3,  # 失败后重试3次
    timeout=10      # 超时时间10秒
)

scheduler.start()
time.sleep(15)
scheduler.stop()
```

## API 文档

### TaskScheduler

任务调度器核心类。

#### 初始化

```python
scheduler = TaskScheduler()
```

#### 方法

##### add_task

添加新任务到调度器。

**参数：**
- `name` (str): 任务名称
- `priority` (int): 优先级（1-10，10最高）
- `execution_time` (datetime): 首次执行时间
- `interval` (int): 执行间隔（秒），0表示不重复
- `func` (callable): 任务执行函数，接收task参数
- `retry_count` (int, 可选): 失败重试次数，默认0
- `timeout` (int, 可选): 执行超时时间（秒），默认300

**返回：**
- `str`: 任务ID

**示例：**
```python
task_id = scheduler.add_task(
    name="我的任务",
    priority=5,
    execution_time=datetime.now(),
    interval=0,
    func=my_function
)
```

##### remove_task

从调度器中删除任务。

**参数：**
- `task_id` (str): 任务ID

**返回：**
- `bool`: 是否成功删除

**示例：**
```python
success = scheduler.remove_task(task_id)
```

##### modify_task_priority

修改任务优先级。

**参数：**
- `task_id` (str): 任务ID
- `new_priority` (int): 新优先级（1-10）

**返回：**
- `bool`: 是否成功修改

**示例：**
```python
success = scheduler.modify_task_priority(task_id, 8)
```

##### get_task_list

获取所有任务列表。

**返回：**
- `List[Dict]`: 任务信息列表

**示例：**
```python
tasks = scheduler.get_task_list()
for task in tasks:
    print(f"{task['name']} - 优先级: {task['priority']}")
```

##### get_task_info

获取特定任务的详细信息。

**参数：**
- `task_id` (str): 任务ID

**返回：**
- `Dict`: 任务详细信息

**示例：**
```python
task_info = scheduler.get_task_info(task_id)
print(task_info)
```

##### start

启动调度器。

**返回：**
- `bool`: 是否成功启动

**示例：**
```python
scheduler.start()
```

##### stop

停止调度器（等待当前任务完成）。

**返回：**
- `bool`: 是否成功停止

**示例：**
```python
scheduler.stop()
```

##### pause

暂停调度器（不调度新任务）。

**返回：**
- `bool`: 是否成功暂停

**示例：**
```python
scheduler.pause()
```

##### resume

恢复调度器。

**返回：**
- `bool`: 是否成功恢复

**示例：**
```python
scheduler.resume()
```

##### get_status

获取调度器状态。

**返回：**
- `Dict`: 包含状态、统计信息等

**示例：**
```python
status = scheduler.get_status()
print(f"状态: {status['state']}")
print(f"总任务数: {status['total_tasks']}")
print(f"已执行: {status['stats']['total_executed']}")
```

## 数据模型

### Task

任务数据模型。

**属性：**
- `task_id` (str): 任务唯一标识符
- `name` (str): 任务名称
- `priority` (int): 优先级（1-10）
- `execution_time` (datetime): 首次执行时间
- `interval` (int): 执行间隔（秒）
- `func` (callable): 任务执行函数
- `retry_count` (int): 失败重试次数
- `timeout` (int): 执行超时时间（秒）
- `state` (TaskState): 任务状态
- `created_at` (datetime): 创建时间
- `last_executed_at` (datetime): 最后执行时间
- `next_execution_time` (datetime): 下次执行时间

### TaskState

任务状态枚举。

- `PENDING`: 待执行
- `RUNNING`: 执行中
- `COMPLETED`: 已完成
- `FAILED`: 已失败

### SchedulerState

调度器状态枚举。

- `STOPPED`: 已停止
- `RUNNING`: 运行中
- `PAUSED`: 已暂停

## 设计架构

项目采用分层架构设计：

```
task_scheduler/
├── __init__.py          # 包导出
├── enums.py             # 枚举类型定义
├── models.py            # 数据模型
├── validator.py         # 数据验证器
├── queue.py             # 任务队列
├── executor.py          # 任务执行器
├── scheduler.py         # 调度器核心
└── logger.py            # 日志系统
```

### 核心组件

1. **TaskScheduler**: 调度器核心，负责协调所有组件
2. **TaskQueue**: 优先级队列，管理待执行任务
3. **TaskExecutor**: 任务执行器，执行任务并处理异常
4. **Task**: 任务实体，封装任务信息
5. **TaskValidator**: 数据验证器，验证任务参数

## 扩展性

项目设计时考虑了扩展性：

1. **持久化接口预留**：Task类支持序列化/反序列化
2. **插件机制**：支持自定义任务执行器和日志处理器
3. **事件钩子**：可扩展任务执行前后的钩子函数

## 注意事项

1. **线程安全**：调度器内部使用锁保护共享资源，但任务执行函数本身需要是线程安全的
2. **异常处理**：任务执行函数应该自行处理异常，或者让异常传播以触发重试机制
3. **资源清理**：任务执行函数应该确保正确释放资源
4. **超时控制**：长时间运行的任务应该设置合理的超时时间

## 许可证

MIT License

## 贡献

欢迎提交 Issue 和 Pull Request！

## 联系方式

如有问题或建议，请提交 Issue。