mctech-mcoor 0.1.0

An orchestrator project for OCR tasks

mc-ocr-orchestrator

mc-ocr-orchestrator 是一个面向 OCR 服务的 Python 中间层客户端封装，用于统一调用文本识别、文本检测、图片分类和目标检测服务。

当前项目通过 config.yml 指向后端服务地址，客户端负责请求封装、结果转换，以及部分异步任务包装。

安装

项目要求 Python 3.12。

uv add --editable ../mc-ocr-orchestrator

安装后可直接导入：

from mcoor import require_clas_client, require_text_client

配置文件

默认使用项目根目录下的 config.yml，也可以在初始化时显式传入。

示例配置：

server:
	text_ocr:
		base_url: "http://localhost"
		port: 8001
		path: ""
	image_clas:
		base_url: "http://localhost"
		port: 8002
		path: ""
	image_det:
		base_url: "http://localhost"
		port: 8003
		path: ""
	timeout: 300

其中：

• text_ocr：文本识别和文本检测服务
• image_clas：图片分类服务
• image_det：目标检测服务
• timeout：HTTP 超时时间，单位秒

基本用法

1. 初始化客户端

使用之前应调用setup方法指定配置文件. 若不指定则使用默认配置

from mcoor import setup
setup("orchestrator_config.yml")

2. 获取客户端并识别

本项目提供三个客户端类, 分别是TextPredictionClient, ClassificationPredictClient, DetectionPredictClient, 用于文本/分类/目标识别. 调用时根据任务场景获取不同的客户端. 客户端实例并不耗费资源, 可以在任意时机获取任意多个.

文本识别示例:

text_finder = require_text_client()
text_boxes: List[TextBox] = text_finder.det_and_rec_d(img_det, img_rec=img_rec, merge_boxes=merge_boxes, directions=[1, 0, 0, 0])

图像分类和目标检测需要传入Sence实例, 用于区分识别场景, 指定要调用的模型和类别枚举.

from mcoor import EnumSence, require_clas_client
class TableIntClasType(IntEnum):
    EMPTY = 0
    TOP_LEFT = 1
    TOP_CENTER = 2
    TOP_RIGHT = 3
sence = EnumSence('table_int_clas', TableIntClasType)
classifier = require_clas_client(sence)
clas_result: List[ClasPrediction] = classifier.predict(tick_pieces)
for r in clas_result:
    print(r.clas.name, r.score)

返回对象

TextBox

文本识别接口通常返回 TextBox 对象，常用字段：

• text：识别文本
• score：置信度
• center_pt：中心点
• points：四点坐标
• chars：字符级结果
• rect：外接矩形

ClasPrediction

• clas：枚举类别
• score：分类分数

DetPrediction

• clas：枚举类别
• score：检测分数
• left / top / right / bottom：框坐标

多线程与并发任务

mcoor.task 提供了一套轻量的任务封装，用于把普通函数调用包装成 Task，再通过线程池并发执行。

这一套接口适合以下场景：

• 同时向多个 OCR 接口发请求
• 同时处理多张图片或多个批次
• 希望先组织任务，再统一执行

核心对象有三个：

• Task：封装一次函数调用
• taskit：把方法声明成 Task 工厂
• gather：使用 ThreadPoolExecutor 并发执行多个 Task

1. 直接执行单个 Task

Task 本质上是把 func + args + kwargs 打包起来，真正执行时调用 run()：

from mcoor.task import Task
def add(self, left: int, right: int) -> int:
    return left + right
task = Task(add, 2, 3)
result = task.run()
print(result)  # 5

2. 使用 gather 并发执行多个任务

gather() 会把多个 Task 提交到线程池中，并按传入顺序返回结果：

from mcoor import Task, gather

sence_table_int = EnumSence('table_int_clas', TableIntClasType)
sence_tunnel_tick = EnumSence('tunnel_tick_clas', TunnelTickType)
classifier = require_clas_client(sence_table_int)
classifier2 = require_clas_client(sence_tunnel_tick)
task1 = Task(classifier.predict, images)
task2 = Task(classifier2.predict, images)

res_table_int, res_tunnel_tick = gather(t1, t2)

可选参数：

• max_workers：线程数，默认等于任务数量
• return_exceptions：是否把异常作为结果返回，默认 False

3. 使用 taskit 注解

在要执行的方法旁添加一个被taskit装饰的 xxx_task() 方法, 就可以将目标方法包装为task, 像执行普通方法一样执行它

from mcoor import Task, taskit, gather


class DemoService:
	def plus(self, left: int, right: int) -> int:
		return left + right

	@taskit("plus")
	def plus_task(self, left: int, right: int) -> Task[..., int]:
		raise NotImplementedError


service = DemoService()
res1, res2 = gather(
    plus_task(1, 2),
    plus_task(2, 3)
)
print(res1)
print(res2)

这里的 plus_task() 自身不实现逻辑，它只负责返回一个 Task，最终执行的仍然是 plus()。

4. 使用本项目提供的现成 *_task() 方法

项目里的客户端已经为常用接口预先提供了任务版本方法，无需自己手写 Task。

例如 TextPredictionClient 提供：

• rec_h_task()
• rec_d_task()
• det_and_rec_h_task()
• det_and_rec_d_task()
• detect_task()

ClassificationPredictClient 提供：

• predict_task()

DetectionPredictClient 提供：

• predict_raw_task()
• predict_resize_task()
• predict_split_task()

文本识别并发示例：

import cv2
from mcoor import gather, require_text_client

text_client = require_text_client()

img1 = cv2.imread("demo_1.png")
img2 = cv2.imread("demo_2.png")

result1, result2 = gather(
    text_client.det_and_rec_h_task(img1),
    text_client.det_and_rec_h_task(img2)
)

for item in result1:
	print(item.text, item.score)

如果你已经提前把一批任务组织好了，也可以统一提交：

tasks = [text_client.det_and_rec_h_task(img) for img in images]
results = gather(*tasks, max_workers=4)

5. 异常处理

默认情况下，只要任意一个任务抛异常，gather() 就会立即抛出异常，并取消尚未开始的任务。

results = gather(*tasks)

如果你希望保留成功结果，并把失败项作为异常对象返回：

results = gather(*tasks, return_exceptions=True)

for item in results:
	if isinstance(item, Exception):
		print("task failed:", item)
	else:
		print("task ok:", item)

6. 使用建议

• 由于python GIL, python并不能开启真正的多线程, 所以cpu密集型的任务即使使用gather也无法加速
• 更适合 I/O 型场景，例如并发调用多个 HTTP OCR 服务
• 任务返回顺序与传入顺序一致，不受完成先后影响
• Task 只是延迟执行封装，创建任务本身不会发请求

运行前说明

本项目本身是个中间层，不直接提供 OCR 推理服务。调用前请确保：

• config.yml 中配置的服务已经启动
• 服务接口路径和端口与配置保持一致

如果服务不可用，调用阶段会抛出 HTTP 或服务端相关异常。