""" JSON数据尿检识别器 — 直接翻译 C++ ProcessJsonDataRequest + Get14Results 完整复制: - UrineCheck_Class::ProcessJsonDataRequest (urinecheck_class.cpp:2052) - UrineCheck14::Get14Results(mapData, X) (UrineCheck14.cpp:1041) - 所有 Red*ResultIndex 修正函数 (UrineCheck14.cpp:1413-1746) - PRO/MAU + SG/PH 关联修正逻辑 Author: 小月 Date: 2026-05-13 """ import math import logging from collections import Counter from typing import Dict, List, Optional, Tuple, Any import numpy as np logger = logging.getLogger("json_recognizer") # 延迟导入（避免循环依赖） _ml_config = None def _get_config(): global _ml_config if _ml_config is None: from ml_config import MLConfigLoader _ml_config = MLConfigLoader() _ml_config.load() return _ml_config # ============================================================================= # 项目类型映射 (C++ getProjectType) # ============================================================================= def get_project_type(paper_type: int) -> int: """paperType → projectType 映射""" mapping = { 1: 10, 2: 2, 5: 2, 3: 1, 4: 1, 6: 9, 7: 9, 8: 9, 9: 9, 10: 7, 11: 7, 12: 7, 13: 7, 14: 8, 15: 8, 16: 8, 17: 8, 18: 8, 19: 8, 20: 20, } return mapping.get(paper_type, 0) # ============================================================================= # 多帧投票 (C++ CountResultsModal) # ============================================================================= def count_results_modal(results: Dict[int, List[int]]) -> Dict[int, int]: """对每个 position，取多帧结果中出现次数最多的 index（众数）""" modal = {} for position, indices in results.items(): if not indices: continue counter = Counter(indices) most_common = counter.most_common(1)[0][0] modal[position] = most_common return modal # ============================================================================= # 结果索引修正函数 (红色试纸, algorithm=0) # 直接翻译 C++ UrineCheck14::GetRed*ResultIndex # ============================================================================= def _safe_round(val) -> int: """safe round like C++ round()""" if isinstance(val, np.ndarray): val = float(val.item()) if val.size == 1 else float(val[0]) return int(round(float(val))) def red_glu_index(hsv_lab, p) -> int: return _safe_round(p) def red_ket_index(hsv_lab, p) -> int: ret = _safe_round(p) h = float(hsv_lab[0]) if ret > 1 and 45 < h < 145: return 0 return ret def red_bil_index(hsv_lab, p) -> int: return _safe_round(p) def red_pro_index(hsv_lab, p) -> int: ret = _safe_round(p) h = float(hsv_lab[0]) if ret > 0 and (h < 25 or h > 110): return 0 return ret def red_nit_index(hsv_lab, p) -> int: ret = _safe_round(p) h = float(hsv_lab[0]) # C++ 20230731 version: h>40 && h<140 if ret > 0 and 40 < h < 140: return 0 return ret def red_ph_index(hsv_lab, p) -> int: return _safe_round(p) def red_bld_index(hsv_lab, p) -> int: return _safe_round(p) def red_sg_index(hsv_lab, p) -> int: return _safe_round(p) def red_wbc_index(hsv_lab, p) -> int: return _safe_round(p) def red_vc_index(hsv_lab, p) -> int: return _safe_round(p) def red_cre_index(hsv_lab, p) -> int: return _safe_round(p) def red_uca_index(hsv_lab, p) -> int: return _safe_round(p) def red_mau_index(hsv_lab, p) -> int: ret = _safe_round(p) h = float(hsv_lab[0]) if ret > 0 and (h < 30 or h > 115): return 0 return ret def red_uro_index(hsv_lab, p) -> int: ret = _safe_round(p) h = float(hsv_lab[0]) if ret > 0 and 25 < h < 150: return 0 return ret # ---- 绿色/蓝色试纸 (algorithm=1) ---- def glu_index(hsv_lab, p) -> int: return _safe_round(p) def ket_index(hsv_lab, p) -> int: return _safe_round(p) def bil_index(hsv_lab, p) -> int: return _safe_round(p) def pro_index(hsv_lab, p) -> int: return _safe_round(p) def nit_index(hsv_lab, p) -> int: return _safe_round(p) def ph_index(hsv_lab, p) -> int: return _safe_round(p) def bld_index(hsv_lab, p) -> int: return _safe_round(p) def sg_index(hsv_lab, p) -> int: return _safe_round(p) def wbc_index(hsv_lab, p) -> int: return _safe_round(p) def vc_index(hsv_lab, p) -> int: return _safe_round(p) def cre_index(hsv_lab, p) -> int: return _safe_round(p) def uca_index(hsv_lab, p) -> int: return _safe_round(p) def mau_index(hsv_lab, p) -> int: return _safe_round(p) def uro_index(hsv_lab, p) -> int: return _safe_round(p) # ---- PET试纸 (algorithm=2) ---- def pet_glu_index(hsv_lab, p) -> int: return _safe_round(p) def pet_sg_index(hsv_lab, p) -> int: return _safe_round(p) # 函数名 → 函数对象映射 RESULT_FUNC_REGISTRY = { # Red (algorithm=0) "red_glu": red_glu_index, "red_ket": red_ket_index, "red_bil": red_bil_index, "red_pro": red_pro_index, "red_nit": red_nit_index, "red_ph": red_ph_index, "red_bld": red_bld_index, "red_sg": red_sg_index, "red_wbc": red_wbc_index, "red_vc": red_vc_index, "red_cre": red_cre_index, "red_uca": red_uca_index, "red_mau": red_mau_index, "red_uro": red_uro_index, # Green/Blue (algorithm=1) "glu": glu_index, "ket": ket_index, "bil": bil_index, "pro": pro_index, "nit": nit_index, "ph": ph_index, "bld": bld_index, "sg": sg_index, "wbc": wbc_index, "vc": vc_index, "cre": cre_index, "uca": uca_index, "mau": mau_index, "uro": uro_index, # PET (algorithm=2) "pet_glu": pet_glu_index, "pet_sg": pet_sg_index, } # ============================================================================= # 核心识别器 # ============================================================================= class UrineJsonRecognizer: """JSON数据尿检识别器""" def __init__(self): self.config = _get_config() def recognize(self, json_body: dict) -> dict: """ 完整识别流程 — 翻译 ProcessJsonDataRequest 输入: 与 txt 文件格式一致的 JSON dict 输出: {"score": "100", "status": "0", "data": [{id, index}, ...]} """ config = self.config # ---- 1. 提取字段 ---- data_array = json_body.get("data", []) extend = json_body.get("extend", {}) if not data_array or len(data_array) == 0: return { "score": "0", "status": "-8", "message": "数据为空！", "project_id": 0, "imgid": 0, } # ---- 2. 确定试纸类型 ---- paper_type = self._extract_paper_type(extend) project_id = get_project_type(paper_type) if paper_type < 1 or paper_type >= 21: return { "score": "0", "status": "-10", "message": "设置试纸类型参数错误！", "project_id": paper_type, "imgid": 0, } X = paper_type - 1 # C++ 中的 papertypeIndex # ---- 3. 逐帧处理 ---- results: Dict[int, List[int]] = {} for frame in data_array: if not isinstance(frame, list): continue # 3a. 解析颜色数据 → mapData map_data = {} for item in frame: item_id = int(item.get("id", 0)) value_str = str(item.get("value", "")) try: values = [float(x) for x in value_str.split(",")] except (ValueError, AttributeError): continue if len(values) >= 6: map_data[item_id] = values[:6] if not map_data: continue # 3b. 调用 ML 推理 frame_result = self._get14_results(map_data, X) # 3c. 收集结果 for position, raw_index in frame_result.items(): if position not in results: results[position] = [] results[position].append(raw_index) if not results: return { "score": "0", "status": "-8", "message": "未能识别任何色块！", "project_id": project_id, "imgid": 0, } # ---- 4. 多帧投票 ---- last_result = count_results_modal(results) # ---- 5. PRO/MAU 关联修正 ---- pro_pos = config.positionsPro[X] mau_pos = config.positionsMau[X] pro_val = -1 mau_val = -1 max_mau = config.get_max_mau(X) if pro_pos >= 0 and mau_pos >= 0: if pro_pos in last_result: pro_val = last_result[pro_pos] if mau_pos in last_result: mau_val = last_result[mau_pos] if mau_val == 0: pro_val = 0 else: if pro_val == 0 and mau_val == max_mau: mau_val = max_mau - 1 if pro_val >= 1: mau_val = max_mau # ---- 6. SG/PH 关联修正 ---- sg_pos = config.positionsSg[X] ph_pos = config.positionsPh[X] sg_val = -1 ph_val = -1 max_sg = config.get_max_sg(X) if sg_pos >= 0 and ph_pos >= 0: if sg_pos in last_result: sg_val = last_result[sg_pos] if ph_pos in last_result: ph_val = last_result[ph_pos] if project_id < 20 and ph_val >= 2 and sg_val < max_sg: sg_val += 1 # ---- 7. 构建输出 data 数组 ---- output_data = [] for position in sorted(last_result.keys()): raw_index = last_result[position] output_id = config.get_stat_id(X, position) # 获取最终 index if pro_pos >= 0 and mau_pos >= 0: if position == pro_pos: raw_index = pro_val elif position == mau_pos: raw_index = mau_val if sg_pos >= 0 and ph_pos >= 0: if position == sg_pos: raw_index = sg_val # 通过 indexmap 映射 output_index = config.get_output_index(X, position, raw_index) output_data.append({ "id": output_id, "index": output_index, }) return { "score": "100", "status": "0", "message": "OK", "project_id": project_id, "paper_id": paper_type, "imgid": 0, "data": output_data, } # ========== ML 推理 ========== def _get14_results(self, map_data: Dict[int, List[float]], paper_type: int) -> Dict[int, int]: """ 完整翻译 UrineCheck14::Get14Results(mapData, X) 对每个模型执行: 取6维特征 → RTrees预测 → 结果索引函数修正 """ config = self.config results = {} for position, model in config.models[paper_type].items(): # 查找该 position 的颜色数据 if position not in map_data: continue vdata = map_data[position] if len(vdata) < 6: continue # 构造 1×6 样本矩阵 sample = np.array([vdata[:6]], dtype=np.float32) # RTrees 预测 try: resp = model.predict(sample) # resp 可能是 (retval, results) 元组或直接值 if isinstance(resp, tuple): p = float(resp[1].item()) if resp[1].size == 1 else float(resp[1][0]) else: p = float(resp) except Exception as e: logger.warning(f"预测失败 paperType={paper_type} pos={position}: {e}") continue # 获取结果索引函数 func_type = config.get_func_type(paper_type, position) if func_type is None: continue func = RESULT_FUNC_REGISTRY.get(func_type) if func is None: continue # 调用结果索引函数 try: pi = func(vdata[:6], p) results[position] = pi except Exception as e: logger.warning(f"索引函数调用失败 {func_type}: {e}") continue return results # ========== 试纸类型提取 ========== def _extract_paper_type(self, extend: dict) -> int: """从 extend 字段提取 paperType""" if not isinstance(extend, dict) or not extend: return 0 # type 优先 if "type" in extend: try: return int(extend["type"]) except (ValueError, TypeError): pass # color_14i 其次 if "color_14i" in extend: try: return int(extend["color_14i"]) except (ValueError, TypeError): pass return 0 # ============================================================================= # 全局单例 # ============================================================================= _recognizer: Optional[UrineJsonRecognizer] = None def get_recognizer() -> UrineJsonRecognizer: global _recognizer if _recognizer is None: _recognizer = UrineJsonRecognizer() return _recognizer