Boen_Shi
6a2e046884
- 新增 analyze、analyze_result、analyze_status 和 health 路由 - 实现图像上传和任务提交功能 - 添加任务状态查询和结果获取接口 - 集成 segformer 和 yolo 模型进行图像检测 - 实现 SAM3 预处理功能用于图像预处理判断 - 添加模型选择配置支持 segformer 和 yolo - 实现任务队列管理和异步处理机制 - 添加 Dockerfile 用于容器化部署 - 配置环境变量和 gitignore 规则 - 创建数据模型定义 API 响应结构
66 lines
1.5 KiB
Python
66 lines
1.5 KiB
Python
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import List, Dict, Optional
|
|
|
|
|
|
class TaskStatus(Enum):
|
|
OK = "正常运行"
|
|
QUEUED = "已入队"
|
|
PROCESSING = "处理中"
|
|
COMPLETED = "处理完成"
|
|
FAILED = "处理错误"
|
|
NOT_FOUND = "任务不存在"
|
|
|
|
|
|
class TaskStore:
|
|
def __init__(self, images: List[str]):
|
|
self._status: str = TaskStatus.QUEUED.name
|
|
self._progress: int = 0
|
|
self._images: List[str] = images
|
|
self._result: List[Dict[str, str]] = []
|
|
self._message: str = ""
|
|
self._completedAt: Optional[datetime] = None
|
|
|
|
@property
|
|
def images(self):
|
|
return self._images
|
|
|
|
@property
|
|
def status(self):
|
|
return self._status
|
|
|
|
@status.setter
|
|
def status(self, status: TaskStatus):
|
|
self._status = status
|
|
|
|
@property
|
|
def progress(self):
|
|
return self._progress
|
|
|
|
@progress.setter
|
|
def progress(self, progress: int):
|
|
self._progress = progress
|
|
|
|
@property
|
|
def result(self):
|
|
return self._result
|
|
|
|
@result.setter
|
|
def result(self, result: str):
|
|
self._result = result
|
|
|
|
@property
|
|
def message(self):
|
|
return self._message
|
|
|
|
@message.setter
|
|
def message(self, message: str):
|
|
self._message = message
|
|
|
|
@property
|
|
def completedAt(self):
|
|
return self._completedAt
|
|
|
|
@completedAt.setter
|
|
def completedAt(self, completedAt: datetime):
|
|
self._completedAt = completedAt |