feat(model-routing): centralize routes in YAML and expose supported l… (#741)

* feat(model-routing): centralize routes in YAML and expose supported languages

* fix(config): correct mobile CLS architecture key casing

* docs: add Backer

* docs: add Backer

* chore: update sponsor way

* feat: add lang_mapping phrase

* refactor: change model cache name

* tests: fixed path error

* tests: add error catch

* fix(tensorrt): resolve precision against hardware capabilities
This commit is contained in:
SWHL authored and GitHub committed 2026-10-08 10:16:32 +08:00
1 parent 332b4fcc23
commit 75946b1d26
12 files changed
+1176 -193

No files matched your search

+5 -5
View File
@@ -12,7 +12,7 @@ from typing import Iterable, List, Optional
PYTHON_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PYTHON_ROOT))
from rapidocr.utils.model_resolver import normalize_lang, resolve_model_key
from rapidocr.utils.model_resolver import normalize_lang, route_to_model_key
from rapidocr.utils.typings import ModelType, OCRVersion, TaskType
DEFAULT_CONFIG_YAML = PYTHON_ROOT / "rapidocr" / "config.yaml"
@@ -152,7 +152,7 @@ def select_model_info(spec: ModelSpec, registry: dict) -> dict:
f"{spec.engine}.{spec.ocr_version}.{spec.task}"
) from e
model_key = _resolve_model_key(spec)
model_key = _route_to_model_key(spec)
if model_key is not None:
model_info = task_models.get(model_key)
if not isinstance(model_info, dict):
@@ -185,7 +185,7 @@ def select_model_info(spec: ModelSpec, registry: dict) -> dict:
)
def _resolve_model_key(spec: ModelSpec) -> Optional[str]:
def _route_to_model_key(spec: ModelSpec) -> Optional[str]:
try:
task_type = TaskType(spec.task)
ocr_version = OCRVersion(spec.ocr_version)
@@ -193,7 +193,7 @@ def _resolve_model_key(spec: ModelSpec) -> Optional[str]:
except ValueError:
return None
return resolve_model_key(task_type, ocr_version, spec.lang, model_type)
return route_to_model_key(task_type, ocr_version, spec.lang, model_type)
def model_info_to_assets(spec: ModelSpec, model_info: dict) -> List[WheelAsset]:
@@ -354,7 +354,7 @@ def _get_content_length(value: Optional[str]) -> Optional[int]:
def _print_progress(downloaded: int, total_size: Optional[int]) -> None:
if not total_size:
return
return
percent = downloaded / total_size * 100
print(
f"\r {downloaded / 1024 / 1024:.1f}MB / "
+108
View File
@@ -0,0 +1,108 @@
# -*- encoding: utf-8 -*-
from collections.abc import Mapping, Sequence
from typing import Any
from omegaconf import OmegaConf
from rapidocr.utils.typings import ModelType, OCRVersion, TaskType
def validate_model_routes(model_routes: Any) -> None:
"""Validate the engine-neutral route schema explicitly."""
if not _is_mapping(model_routes):
raise ValueError("Invalid model_routes: expected a mapping")
for path, routes in _iter_model_route_lists(model_routes):
_validate_model_route_list(path, routes)
def _iter_model_route_lists(model_routes: Any):
for version, version_cfg in model_routes.items():
if version == "_defs":
continue
_validate_enum("OCR version", version, OCRVersion)
if not _is_mapping(version_cfg):
raise ValueError(f"Invalid model route version: {version}")
if "aliases" in version_cfg:
_validate_aliases(f"{version}.aliases", version_cfg["aliases"])
for task, task_cfg in version_cfg.items():
if task == "aliases":
continue
_validate_enum("task", task, TaskType)
if not _is_mapping(task_cfg):
raise ValueError(f"Invalid model route task: {version}.{task}")
if "aliases" in task_cfg:
_validate_aliases(f"{version}.{task}.aliases", task_cfg["aliases"])
for model_type, routes in task_cfg.items():
if model_type != "aliases":
_validate_enum("model type", model_type, ModelType)
yield f"{version}.{task}.{model_type}", routes
def _validate_enum(label: str, value: str, enum_type: Any) -> None:
try:
enum_type(value)
except ValueError as exc:
valid_values = ", ".join(item.value for item in enum_type)
raise ValueError(
f"Invalid {label} {value!r}; expected one of: {valid_values}"
) from exc
def _is_mapping(value: Any) -> bool:
return OmegaConf.is_dict(value) or isinstance(value, Mapping)
def _is_sequence(value: Any) -> bool:
return OmegaConf.is_list(value) or (
isinstance(value, Sequence) and not isinstance(value, (str, bytes))
)
def _validate_aliases(path: str, aliases: Any) -> None:
if not _is_mapping(aliases):
raise ValueError(f"Invalid aliases: {path}")
for alias, target in aliases.items():
if not isinstance(alias, str) or not alias.strip():
raise ValueError(f"Invalid alias: {path}")
if not isinstance(target, str) or not target.strip():
raise ValueError(f"Invalid alias target: {path}.{alias}")
def _validate_model_route_list(path: str, routes: Any) -> None:
if not _is_sequence(routes):
raise ValueError(f"Invalid model route list: {path}")
seen = set()
seen_languages = set()
for route in routes:
if not _is_mapping(route):
raise ValueError(f"Invalid model route entry: {path}")
required = {"lang", "model_key", "supported_langs"}
if not required.issubset(route):
raise ValueError(f"Invalid model route entry: {path}")
if not isinstance(route["lang"], str) or not route["lang"].strip():
raise ValueError(f"Invalid route language: {path}")
model_key = route["model_key"]
if not isinstance(model_key, str) or not model_key.strip():
raise ValueError(f"Invalid route model key: {path}")
if model_key in seen:
raise ValueError(f"Duplicate model route key: {model_key}")
seen.add(model_key)
supported_langs = route["supported_langs"]
if not _is_sequence(supported_langs) or not supported_langs:
raise ValueError(f"Empty supported languages: {model_key}")
for lang in supported_langs:
if not isinstance(lang, str) or not lang.strip():
raise ValueError(f"Invalid supported language: {model_key}")
if lang in seen_languages:
raise ValueError(f"Overlapping supported language: {path}.{lang}")
seen_languages.add(lang)