Spaces:
Sleeping
Sleeping
| """Chargement et inférence du modèle CRNN fine-tuné sur les écrans de pompe. | |
| C'est le "modèle IA développé" au sens du §8.4 du cahier des charges : | |
| un CRNN (crnn_vgg16_bn, doctr) fine-tuné sur nos propres données annotées | |
| (voir train/) | |
| """ | |
| import os | |
| from typing import Any, Dict, List, Optional, Tuple | |
| import cv2 | |
| import numpy as np | |
| import torch | |
| import torch.nn.functional as F | |
| from .preprocessing import FIELD_NAMES, split_lcd_lines | |
| try: | |
| from doctr.datasets import VOCABS | |
| from doctr.models import crnn_vgg16_bn | |
| except Exception as exc: # pragma: no cover - dépend de l'environnement | |
| VOCABS = None | |
| crnn_vgg16_bn = None | |
| _DOCTR_IMPORT_ERROR = exc | |
| else: | |
| _DOCTR_IMPORT_ERROR = None | |
| VOCAB = VOCABS["french"] if VOCABS is not None else None | |
| IMG_H = 32 | |
| MODULE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| DEFAULT_MODEL_PATH = os.path.join(MODULE_ROOT, "models", "crnn_fuel_pump_best.pt") | |
| def get_device() -> torch.device: | |
| return torch.device("cuda" if torch.cuda.is_available() else "cpu") | |
| def load_model(model_path: Optional[str] = None, device: Optional[torch.device] = None): | |
| if model_path is None: | |
| model_path = DEFAULT_MODEL_PATH | |
| if not os.path.exists(model_path): | |
| raise FileNotFoundError(f"Modèle introuvable : {model_path}") | |
| if crnn_vgg16_bn is None or VOCAB is None: | |
| raise ImportError( | |
| "python-doctr n'est pas installé (voir requirements.txt)." | |
| ) from _DOCTR_IMPORT_ERROR | |
| if device is None: | |
| device = get_device() | |
| model = crnn_vgg16_bn(pretrained=False, pretrained_backbone=False, vocab=VOCAB) | |
| model.load_state_dict(torch.load(model_path, map_location=device)) | |
| return model.to(device).eval() | |
| def _resize_preserve_aspect(tensor: torch.Tensor, img_h: int) -> torch.Tensor: | |
| """Redimensionne en conservant le ratio d'aspect (hauteur fixe). Doit | |
| rester cohérent avec le prétraitement utilisé à l'entraînement | |
| (train/finetune_doctr.py::_resize_preserve_aspect) : un écart entre les | |
| deux dégrade silencieusement la précision du modèle. | |
| """ | |
| _, h, w = tensor.shape | |
| new_w = max(16, int(img_h * w / max(h, 1))) | |
| resized = F.interpolate( | |
| tensor.unsqueeze(0), size=(img_h, new_w), | |
| mode="bilinear", align_corners=False, | |
| ).squeeze(0) | |
| mn, mx = resized.min(), resized.max() | |
| if mx - mn > 0.01: | |
| resized = (resized - mn) / (mx - mn) | |
| return resized.unsqueeze(0) | |
| def preprocess_crop(crop_bgr: np.ndarray, img_h: int = IMG_H) -> torch.Tensor: | |
| img = cv2.cvtColor(crop_bgr, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0 | |
| tensor = torch.from_numpy(img).permute(2, 0, 1) | |
| return _resize_preserve_aspect(tensor, img_h) | |
| def predict_line(model, crop_bgr: np.ndarray, device: torch.device) -> Tuple[str, float]: | |
| tensor = preprocess_crop(crop_bgr) | |
| out = model(tensor.to(device), return_preds=True) | |
| preds = out.get("preds", []) | |
| if not preds: | |
| return "", 0.0 | |
| text, confidence = preds[0] | |
| return text, max(float(confidence), 0.0) | |
| def recognize_screen(model, screen_crop: np.ndarray, device: torch.device, | |
| n_lines: int = 3) -> List[Dict[str, Any]]: | |
| """Découpe le crop d'écran en lignes puis lit chaque valeur avec le CRNN. | |
| Retourne une liste de {'field', 'text', 'confidence'} — un élément par | |
| ligne détectée (prix / volume / prix_litre), dans l'ordre d'affichage. | |
| """ | |
| line_crops, _ = split_lcd_lines(screen_crop, n_lines=n_lines) | |
| results = [] | |
| for idx, crop in enumerate(line_crops): | |
| if crop.size == 0: | |
| continue | |
| text, confidence = predict_line(model, crop, device) | |
| field = FIELD_NAMES[idx] if idx < len(FIELD_NAMES) else f"ligne_{idx}" | |
| results.append({"field": field, "text": text, "confidence": confidence}) | |
| return results | |