-
-
Save SWHL/fd6e647337346aa3e268887a4bf179e6 to your computer and use it in GitHub Desktop.
RapidOCR 测试 PyTorch MPS benchmark 代码
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python | |
| # -*- encoding: utf-8 -*- | |
| """ | |
| Benchmark script for comparing PyTorch CPU vs MPS | |
| Detection models only | |
| """ | |
| import traceback | |
| from typing import Any, Dict, List | |
| import cv2 | |
| import numpy as np | |
| from datasets import load_dataset | |
| from tqdm import tqdm | |
| from rapidocr import RapidOCR | |
| from rapidocr.utils.typings import EngineType, ModelType, OCRVersion | |
| # Load dataset | |
| print("Loading dataset...") | |
| dataset = load_dataset("SWHL/text_det_test_dataset") | |
| test_data = list(dataset["test"]) | |
| print(f"Dataset loaded with {len(test_data)} samples") | |
| def benchmark_det(engine: RapidOCR, engine_name: str) -> Dict[str, Any]: | |
| """ | |
| Run benchmark on detection model | |
| Args: | |
| engine: RapidOCR instance | |
| engine_name: Name of the engine configuration | |
| Returns: | |
| Dictionary with benchmark results | |
| """ | |
| content = [] | |
| total_elapse = 0.0 | |
| print(f"\nRunning benchmark for: {engine_name}") | |
| for one_data in tqdm(test_data, desc=f"Processing {engine_name}"): | |
| img = np.array(one_data.get("image")) | |
| img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) | |
| ocr_results = engine(img, use_det=True, use_cls=False, use_rec=False) | |
| dt_boxes = ocr_results.boxes | |
| dt_boxes = [] if dt_boxes is None else dt_boxes.tolist() | |
| elapse = ocr_results.elapse if ocr_results.elapse is not None else 0.0 | |
| total_elapse += elapse | |
| gt_boxes = [v["points"] for v in one_data["shapes"]] | |
| content.append(f"{dt_boxes}\t{gt_boxes}\t{elapse}") | |
| # Save predictions | |
| pred_file = f"pred_{engine_name.replace(' ', '_').replace('/', '_')}.txt" | |
| with open(pred_file, "w", encoding="utf-8") as f: | |
| for v in content: | |
| f.write(f"{v}\n") | |
| # Calculate metrics | |
| try: | |
| from text_det_metric import TextDetMetric | |
| metric = TextDetMetric() | |
| metric_results = metric(pred_file) | |
| return { | |
| "engine_name": engine_name, | |
| "precision": metric_results.get("precision", 0.0), | |
| "recall": metric_results.get("recall", 0.0), | |
| "hmean": metric_results.get("hmean", 0.0), | |
| "avg_elapse": total_elapse / len(test_data), | |
| "total_elapse": total_elapse, | |
| "pred_file": pred_file, | |
| } | |
| except ImportError: | |
| print("Warning: text_det_metric not found, skipping metric calculation") | |
| return { | |
| "engine_name": engine_name, | |
| "precision": None, | |
| "recall": None, | |
| "hmean": None, | |
| "avg_elapse": total_elapse / len(test_data), | |
| "total_elapse": total_elapse, | |
| "pred_file": pred_file, | |
| } | |
| # Engine configurations to test | |
| ENGINE_CONFIGS = [ | |
| # PyTorch CPU - PP-OCRv4 | |
| { | |
| "name": "PP-OCRv4 Det / PyTorch / CPU", | |
| "config": { | |
| "Global.use_det": True, | |
| "Global.use_cls": False, | |
| "Global.use_rec": False, | |
| "Det.engine_type": EngineType.TORCH, | |
| "Det.ocr_version": OCRVersion.PPOCRV4, | |
| "EngineConfig.torch.use_mps": False, | |
| }, | |
| }, | |
| # PyTorch MPS - PP-OCRv4 | |
| { | |
| "name": "PP-OCRv4 Det / PyTorch / MPS", | |
| "config": { | |
| "Global.use_det": True, | |
| "Global.use_cls": False, | |
| "Global.use_rec": False, | |
| "Det.engine_type": EngineType.TORCH, | |
| "Det.ocr_version": OCRVersion.PPOCRV4, | |
| "EngineConfig.torch.use_mps": True, | |
| }, | |
| }, | |
| # PyTorch CPU - PP-OCRv5 | |
| { | |
| "name": "PP-OCRv5 Det Mobile / PyTorch / CPU", | |
| "config": { | |
| "Global.use_det": True, | |
| "Global.use_cls": False, | |
| "Global.use_rec": False, | |
| "Det.engine_type": EngineType.TORCH, | |
| "Det.ocr_version": OCRVersion.PPOCRV5, | |
| "Det.model_type": ModelType.MOBILE, | |
| "EngineConfig.torch.use_mps": False, | |
| }, | |
| }, | |
| # PyTorch MPS - PP-OCRv5 | |
| { | |
| "name": "PP-OCRv5 Det Mobile / PyTorch / MPS", | |
| "config": { | |
| "Global.use_det": True, | |
| "Global.use_cls": False, | |
| "Global.use_rec": False, | |
| "Det.engine_type": EngineType.TORCH, | |
| "Det.ocr_version": OCRVersion.PPOCRV5, | |
| "Det.model_type": ModelType.MOBILE, | |
| "EngineConfig.torch.use_mps": True, | |
| }, | |
| }, | |
| ] | |
| def generate_markdown_table(results: List[Dict[str, Any]]) -> str: | |
| """Generate markdown table from benchmark results""" | |
| # Table header | |
| table = "| Exp | 模型 | 推理框架 | 推理引擎 | Precision↑ | Recall↑ | H-mean↑ | Elapse↓(ms) |\n" | |
| table += "|-----|------|----------|----------|------------|---------|---------|-------------|\n" | |
| # Table rows | |
| for i, result in enumerate(results, 1): | |
| # Parse name | |
| parts = result["engine_name"].split(" / ") | |
| model = parts[0] if len(parts) > 0 else "Unknown" | |
| framework = parts[1] if len(parts) > 1 else "Unknown" | |
| engine = parts[2] if len(parts) > 2 else "Unknown" | |
| precision = ( | |
| f"{result['precision']:.4f}" if result["precision"] is not None else "N/A" | |
| ) | |
| recall = f"{result['recall']:.4f}" if result["recall"] is not None else "N/A" | |
| hmean = f"{result['hmean']:.4f}" if result["hmean"] is not None else "N/A" | |
| elapse = f"{result['avg_elapse'] * 1000:.2f}" | |
| table += f"| {i} | {model} | {framework} | {engine} | {precision} | {recall} | {hmean} | {elapse} |\n" | |
| return table | |
| def main(): | |
| results = [] | |
| for engine_config in ENGINE_CONFIGS: | |
| print("\n" + "=" * 100) | |
| print(f"Testing: {engine_config['name']}") | |
| print("=" * 100) | |
| try: | |
| engine = RapidOCR(params=engine_config["config"]) | |
| result = benchmark_det(engine, engine_config["name"]) | |
| results.append(result) | |
| print(f"\nResults for {engine_config['name']}:") | |
| print(f" Precision: {result['precision']}") | |
| print(f" Recall: {result['recall']}") | |
| print(f" H-mean: {result['hmean']}") | |
| print(f" Avg Elapse: {result['avg_elapse'] * 1000:.2f} ms") | |
| except Exception: | |
| print(f"\nERROR testing {engine_config['name']}:") | |
| print(traceback.format_exc()) | |
| print("=" * 100) | |
| continue | |
| # Generate and save markdown table | |
| if results: | |
| print("\n" + "=" * 100) | |
| print("BENCHMARK RESULTS - DETECTION MODELS") | |
| print("=" * 100) | |
| markdown_table = generate_markdown_table(results) | |
| print("\n" + markdown_table) | |
| # Save to file | |
| output_file = "benchmark_torch_mps_det_results.md" | |
| with open(output_file, "w", encoding="utf-8") as f: | |
| f.write("# PyTorch MPS vs CPU Benchmark Results - Detection Models\n\n") | |
| f.write("## Detection Models\n\n") | |
| f.write(markdown_table) | |
| f.write("\n\n## Notes\n\n") | |
| f.write("- Precision↑: Higher is better\n") | |
| f.write("- Recall↑: Higher is better\n") | |
| f.write( | |
| "- H-mean↑: Harmonic mean of precision and recall, higher is better\n" | |
| ) | |
| f.write( | |
| "- Elapse↓: Average inference time per image in milliseconds, lower is better\n" | |
| ) | |
| print(f"\nResults saved to: {output_file}") | |
| else: | |
| print("\nNo results to report!") | |
| if __name__ == "__main__": | |
| main() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python | |
| # -*- encoding: utf-8 -*- | |
| """ | |
| Benchmark script for comparing PyTorch CPU vs MPS | |
| Recognition models only | |
| """ | |
| import time | |
| import traceback | |
| from typing import Any, Dict, List | |
| import cv2 | |
| import numpy as np | |
| from datasets import load_dataset | |
| from tqdm import tqdm | |
| from rapidocr import RapidOCR | |
| from rapidocr.utils.typings import EngineType, ModelType, OCRVersion | |
| # Load dataset | |
| print("Loading recognition dataset...") | |
| dataset = load_dataset("SWHL/text_rec_test_dataset") | |
| test_data = list(dataset["test"]) | |
| print(f"Dataset loaded with {len(test_data)} samples") | |
| def benchmark_rec(engine: RapidOCR, engine_name: str) -> Dict[str, Any]: | |
| """ | |
| Run benchmark on recognition model | |
| Args: | |
| engine: RapidOCR instance | |
| engine_name: Name of the engine configuration | |
| Returns: | |
| Dictionary with benchmark results | |
| """ | |
| content = [] | |
| total_elapse = 0.0 | |
| print(f"\nRunning benchmark for: {engine_name}") | |
| for one_data in tqdm(test_data, desc=f"Processing {engine_name}"): | |
| img = np.array(one_data.get("image")) | |
| img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) | |
| t0 = time.perf_counter() | |
| result = engine(img, use_rec=True, use_cls=False, use_det=False) | |
| elapse = time.perf_counter() - t0 | |
| rec_text = result.txts[0] if result.txts and len(result.txts) > 0 else "" | |
| if len(rec_text) <= 0: | |
| rec_text = "" | |
| elapse = 0.0 | |
| total_elapse += elapse | |
| gt = one_data.get("label", "") | |
| content.append(f"{rec_text}\t{gt}\t{elapse}") | |
| # Save predictions | |
| pred_file = f"pred_rec_{engine_name.replace(' ', '_').replace('/', '_')}.txt" | |
| with open(pred_file, "w", encoding="utf-8") as f: | |
| for v in content: | |
| f.write(f"{v}\n") | |
| # Calculate metrics | |
| try: | |
| from text_rec_metric import TextRecMetric | |
| metric = TextRecMetric() | |
| metric_results = metric(pred_file) | |
| return { | |
| "engine_name": engine_name, | |
| "exact_match": metric_results.get("ExactMatch", 0.0), | |
| "char_match": metric_results.get("CharMatch", 0.0), | |
| "avg_elapse": total_elapse / len(test_data), | |
| "total_elapse": total_elapse, | |
| "pred_file": pred_file, | |
| } | |
| except ImportError: | |
| print("Warning: text_rec_metric not found, skipping metric calculation") | |
| return { | |
| "engine_name": engine_name, | |
| "exact_match": None, | |
| "char_match": None, | |
| "avg_elapse": total_elapse / len(test_data), | |
| "total_elapse": total_elapse, | |
| "pred_file": pred_file, | |
| } | |
| # Engine configurations to test | |
| ENGINE_CONFIGS = [ | |
| # PP-OCRv4 Rec Mobile - PyTorch CPU | |
| { | |
| "name": "PP-OCRv4 Rec Mobile / PyTorch / CPU", | |
| "config": { | |
| "Global.use_det": False, | |
| "Global.use_cls": False, | |
| "Global.use_rec": True, | |
| "Rec.engine_type": EngineType.TORCH, | |
| "Rec.ocr_version": OCRVersion.PPOCRV4, | |
| "EngineConfig.torch.use_mps": False, | |
| }, | |
| }, | |
| # PP-OCRv4 Rec Mobile - PyTorch MPS | |
| { | |
| "name": "PP-OCRv4 Rec Mobile / PyTorch / MPS", | |
| "config": { | |
| "Global.use_det": False, | |
| "Global.use_cls": False, | |
| "Global.use_rec": True, | |
| "Rec.engine_type": EngineType.TORCH, | |
| "Rec.ocr_version": OCRVersion.PPOCRV4, | |
| "EngineConfig.torch.use_mps": True, | |
| }, | |
| }, | |
| # PP-OCRv5 Rec Mobile - PyTorch CPU | |
| { | |
| "name": "PP-OCRv5 Rec Mobile / PyTorch / CPU", | |
| "config": { | |
| "Global.use_det": False, | |
| "Global.use_cls": False, | |
| "Global.use_rec": True, | |
| "Rec.engine_type": EngineType.TORCH, | |
| "Rec.ocr_version": OCRVersion.PPOCRV5, | |
| "Rec.model_type": ModelType.MOBILE, | |
| "EngineConfig.torch.use_mps": False, | |
| }, | |
| }, | |
| # PP-OCRv5 Rec Mobile - PyTorch MPS | |
| { | |
| "name": "PP-OCRv5 Rec Mobile / PyTorch / MPS", | |
| "config": { | |
| "Global.use_det": False, | |
| "Global.use_cls": False, | |
| "Global.use_rec": True, | |
| "Rec.engine_type": EngineType.TORCH, | |
| "Rec.ocr_version": OCRVersion.PPOCRV5, | |
| "Rec.model_type": ModelType.MOBILE, | |
| "EngineConfig.torch.use_mps": True, | |
| }, | |
| }, | |
| ] | |
| def generate_markdown_table(results: List[Dict[str, Any]]) -> str: | |
| """Generate markdown table from benchmark results""" | |
| # Table header | |
| table = "| Exp | 模型 | 推理框架 | 推理引擎 | ExactMatch↑ | CharMatch↑ | Elapse↓(ms) |\n" | |
| table += "|-----|------|----------|----------|-------------|------------|-------------|\n" | |
| # Table rows | |
| for i, result in enumerate(results, 1): | |
| # Parse name | |
| parts = result["engine_name"].split(" / ") | |
| model = parts[0] if len(parts) > 0 else "Unknown" | |
| framework = parts[1] if len(parts) > 1 else "Unknown" | |
| engine = parts[2] if len(parts) > 2 else "Unknown" | |
| exact_match = ( | |
| f"{result['exact_match']:.4f}" | |
| if result["exact_match"] is not None | |
| else "N/A" | |
| ) | |
| char_match = ( | |
| f"{result['char_match']:.4f}" if result["char_match"] is not None else "N/A" | |
| ) | |
| elapse = f"{result['avg_elapse'] * 1000:.2f}" | |
| table += f"| {i} | {model} | {framework} | {engine} | {exact_match} | {char_match} | {elapse} |\n" | |
| return table | |
| def main(): | |
| results = [] | |
| for engine_config in ENGINE_CONFIGS: | |
| print("\n" + "=" * 100) | |
| print(f"Testing: {engine_config['name']}") | |
| print("=" * 100) | |
| try: | |
| engine = RapidOCR(params=engine_config["config"]) | |
| result = benchmark_rec(engine, engine_config["name"]) | |
| results.append(result) | |
| print(f"\nResults for {engine_config['name']}:") | |
| print(f" ExactMatch: {result['exact_match']}") | |
| print(f" CharMatch: {result['char_match']}") | |
| print(f" Avg Elapse: {result['avg_elapse'] * 1000:.2f} ms") | |
| except Exception: | |
| print(f"\nERROR testing {engine_config['name']}:") | |
| print(traceback.format_exc()) | |
| print("=" * 100) | |
| continue | |
| # Generate and save markdown table | |
| if results: | |
| print("\n" + "=" * 100) | |
| print("BENCHMARK RESULTS - RECOGNITION MODELS") | |
| print("=" * 100) | |
| markdown_table = generate_markdown_table(results) | |
| print("\n" + markdown_table) | |
| # Save to file | |
| output_file = "benchmark_torch_mps_rec_results.md" | |
| with open(output_file, "w", encoding="utf-8") as f: | |
| f.write( | |
| "# PyTorch MPS vs CPU Benchmark Results - Recognition Models\n\n" | |
| ) | |
| f.write("## Recognition Models\n\n") | |
| f.write(markdown_table) | |
| f.write("\n\n## Notes\n\n") | |
| f.write("- ExactMatch↑: Exact match accuracy, higher is better\n") | |
| f.write("- CharMatch↑: Character-level match accuracy, higher is better\n") | |
| f.write( | |
| "- Elapse↓: Average inference time per image in milliseconds, lower is better\n" | |
| ) | |
| print(f"\nResults saved to: {output_file}") | |
| else: | |
| print("\nNo results to report!") | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment