diff --git a/unilabos/devices/virtual/virtual_sample_demo.py b/unilabos/devices/virtual/virtual_sample_demo.py new file mode 100644 index 00000000..5d85e0f7 --- /dev/null +++ b/unilabos/devices/virtual/virtual_sample_demo.py @@ -0,0 +1,88 @@ +"""虚拟样品演示设备 — 用于前端 sample tracking 功能的极简 demo""" + +import asyncio +import logging +import random +import time +from typing import Any, Dict, List, Optional + + +class VirtualSampleDemo: + """虚拟样品追踪演示设备，提供两种典型返回模式： + - measure_samples: 等长输入输出 (前端按 index 自动对齐) + - split_and_measure: 输出比输入长，附带 samples 列标注归属 + """ + + def __init__(self, device_id: Optional[str] = None, config: Optional[Dict[str, Any]] = None, **kwargs): + if device_id is None and "id" in kwargs: + device_id = kwargs.pop("id") + if config is None and "config" in kwargs: + config = kwargs.pop("config") + + self.device_id = device_id or "unknown_sample_demo" + self.config = config or {} + self.logger = logging.getLogger(f"VirtualSampleDemo.{self.device_id}") + self.data: Dict[str, Any] = {"status": "Idle"} + + # ------------------------------------------------------------------ + # Action 1: 等长输入输出，无 samples 列 + # ------------------------------------------------------------------ + async def measure_samples(self, concentrations: List[float]) -> Dict[str, Any]: + """模拟光度测量。absorbance = concentration * 0.05 + noise + + 入参和出参 list 长度相等，前端按 index 自动对齐。 + """ + self.logger.info(f"measure_samples: concentrations={concentrations}") + absorbance = [round(c * 0.05 + random.gauss(0, 0.005), 4) for c in concentrations] + return {"concentrations": concentrations, "absorbance": absorbance} + + # ------------------------------------------------------------------ + # Action 2: 输出比输入长，带 samples 列 + # ------------------------------------------------------------------ + async def split_and_measure(self, volumes: List[float], split_count: int = 3) -> Dict[str, Any]: + """将每个样品均分为 split_count 份后逐份测量。 + + 返回的 list 长度 = len(volumes) * split_count， + 附带 samples 列标注每行属于第几个输入样品 (0-based index)。 + """ + self.logger.info(f"split_and_measure: volumes={volumes}, split_count={split_count}") + out_volumes: List[float] = [] + readings: List[float] = [] + samples: List[int] = [] + + for idx, vol in enumerate(volumes): + split_vol = round(vol / split_count, 2) + for _ in range(split_count): + out_volumes.append(split_vol) + readings.append(round(random.uniform(0.1, 1.0), 4)) + samples.append(idx) + + return {"volumes": out_volumes, "readings": readings, "samples": samples} + + # ------------------------------------------------------------------ + # Action 3: 入参和出参都带 samples 列（不等长） + # ------------------------------------------------------------------ + async def analyze_readings(self, readings: List[float], samples: List[int]) -> Dict[str, Any]: + """对 split_and_measure 的输出做二次分析。 + + 入参 readings/samples 长度相同但 > 原始样品数， + 出参同样带 samples 列，长度与入参一致。 + """ + self.logger.info(f"analyze_readings: readings={readings}, samples={samples}") + scores: List[float] = [] + passed: List[bool] = [] + threshold = 0.4 + + for r in readings: + score = round(r * 100 + random.gauss(0, 2), 2) + scores.append(score) + passed.append(r >= threshold) + + return {"scores": scores, "passed": passed, "samples": samples} + + # ------------------------------------------------------------------ + # 状态属性 + # ------------------------------------------------------------------ + @property + def status(self) -> str: + return self.data.get("status", "Idle") diff --git a/unilabos/registry/devices/virtual_device.yaml b/unilabos/registry/devices/virtual_device.yaml index 67560f2f..527e98f7 100644 --- a/unilabos/registry/devices/virtual_device.yaml +++ b/unilabos/registry/devices/virtual_device.yaml @@ -2804,6 +2804,294 @@ virtual_rotavap: - vacuum_pressure type: object version: 1.0.0 +virtual_sample_demo: + category: + - virtual_device + class: + action_value_mappings: + analyze_readings: + feedback: {} + goal: + readings: readings + samples: samples + goal_default: + readings: [] + samples: [] + handles: + input: + - data_key: readings + data_source: handle + data_type: sample_list + handler_key: readings_in + label: 测量读数 + - data_key: samples + data_source: handle + data_type: sample_index + handler_key: samples_in + label: 样品索引 + output: + - data_key: scores + data_source: executor + data_type: sample_list + handler_key: scores_out + label: 分析得分 + - data_key: passed + data_source: executor + data_type: sample_list + handler_key: passed_out + label: 是否通过 + - data_key: samples + data_source: executor + data_type: sample_index + handler_key: samples_result_out + label: 样品索引 + placeholder_keys: {} + result: + passed: passed + samples: samples + scores: scores + schema: + description: 对 split_and_measure 输出做二次分析，入参和出参都带 samples 列 + properties: + feedback: + properties: {} + required: [] + title: AnalyzeReadings_Feedback + type: object + goal: + properties: + readings: + description: 测量读数（来自 split_and_measure） + items: + type: number + type: array + samples: + description: 每行归属的输入样品 index (0-based) + items: + type: integer + type: array + required: + - readings + - samples + title: AnalyzeReadings_Goal + type: object + result: + properties: + passed: + description: 是否通过阈值 + items: + type: boolean + type: array + samples: + description: 每行归属的输入样品 index (0-based) + items: + type: integer + type: array + scores: + description: 分析得分 + items: + type: number + type: array + required: + - scores + - passed + - samples + title: AnalyzeReadings_Result + type: object + required: + - goal + title: AnalyzeReadings + type: object + type: UniLabJsonCommandAsync + auto-cleanup: + feedback: {} + goal: {} + goal_default: {} + handles: {} + placeholder_keys: {} + result: {} + schema: + description: cleanup的参数schema + properties: + feedback: {} + goal: + properties: {} + required: [] + type: object + result: {} + required: + - goal + title: cleanup参数 + type: object + type: UniLabJsonCommandAsync + measure_samples: + feedback: {} + goal: + concentrations: concentrations + goal_default: + concentrations: [] + handles: + output: + - data_key: concentrations + data_source: executor + data_type: sample_list + handler_key: concentrations_out + label: 浓度列表 + - data_key: absorbance + data_source: executor + data_type: sample_list + handler_key: absorbance_out + label: 吸光度列表 + placeholder_keys: {} + result: + absorbance: absorbance + concentrations: concentrations + schema: + description: 模拟光度测量，入参出参等长 + properties: + feedback: + properties: {} + required: [] + title: MeasureSamples_Feedback + type: object + goal: + properties: + concentrations: + description: 样品浓度列表 + items: + type: number + type: array + required: + - concentrations + title: MeasureSamples_Goal + type: object + result: + properties: + absorbance: + description: 吸光度列表（与浓度等长） + items: + type: number + type: array + concentrations: + description: 原始浓度列表 + items: + type: number + type: array + required: + - concentrations + - absorbance + title: MeasureSamples_Result + type: object + required: + - goal + title: MeasureSamples + type: object + type: UniLabJsonCommandAsync + split_and_measure: + feedback: {} + goal: + split_count: split_count + volumes: volumes + goal_default: + split_count: 3 + volumes: [] + handles: + output: + - data_key: readings + data_source: executor + data_type: sample_list + handler_key: readings_out + label: 测量读数 + - data_key: samples + data_source: executor + data_type: sample_index + handler_key: samples_out + label: 样品索引 + - data_key: volumes + data_source: executor + data_type: sample_list + handler_key: volumes_out + label: 均分体积 + placeholder_keys: {} + result: + readings: readings + samples: samples + volumes: volumes + schema: + description: 均分样品后逐份测量，输出带 samples 列标注归属 + properties: + feedback: + properties: {} + required: [] + title: SplitAndMeasure_Feedback + type: object + goal: + properties: + split_count: + description: 每个样品均分的份数 + type: integer + volumes: + description: 样品体积列表 + items: + type: number + type: array + required: + - volumes + title: SplitAndMeasure_Goal + type: object + result: + properties: + readings: + description: 测量读数 + items: + type: number + type: array + samples: + description: 每行归属的输入样品 index (0-based) + items: + type: integer + type: array + volumes: + description: 均分后的体积列表 + items: + type: number + type: array + required: + - volumes + - readings + - samples + title: SplitAndMeasure_Result + type: object + required: + - goal + title: SplitAndMeasure + type: object + type: UniLabJsonCommandAsync + module: unilabos.devices.virtual.virtual_sample_demo:VirtualSampleDemo + status_types: + status: str + type: python + config_info: [] + description: Virtual sample tracking demo device + handles: [] + icon: '' + init_param_schema: + config: + properties: + config: + type: string + device_id: + type: string + required: [] + type: object + data: + properties: + status: + type: string + required: + - status + type: object + version: 1.0.0 virtual_separator: category: - virtual_device diff --git a/unilabos/utils/type_check.py b/unilabos/utils/type_check.py index e3df2dc2..074ae91b 100644 --- a/unilabos/utils/type_check.py +++ b/unilabos/utils/type_check.py @@ -82,7 +82,7 @@ def get_result_info_str(error: str, suc: bool, return_value=None) -> str: """ samples = None if isinstance(return_value, dict): - if "samples" in return_value: + if "samples" in return_value and type(return_value["samples"]) in [list, tuple] and type(return_value["samples"][0]) == dict: samples = return_value.pop("samples") result_info = {"error": error, "suc": suc, "return_value": return_value, "samples": samples}