1031

unilabos/devices/workstation/bioyond_studio/bioyond_cell/bioyond_cell_workstation.py

@@ -253,7 +253,7 @@ class BioyondCellWorkstation(BioyondWorkstation):
     def auto_feeding4to3(
         self,
         # ★ 修改点：默认模板路径
-        xlsx_path: Optional[str] = "/Users/calvincao/Desktop/work/uni-lab-all/Uni-Lab-OS/unilabos/devices/workstation/bioyond_studio/bioyond_cell/material_template.xlsx",
+        xlsx_path: Optional[str] = "unilabos\\devices\\workstation\\bioyond_studio\\bioyond_cell\\material_template.xlsx",
         # ---------------- WH4 - 加样头面 (Z=1, 12个点位) ----------------
         WH4_x1_y1_z1_1_materialName: str = "", WH4_x1_y1_z1_1_quantity: float = 0.0,
         WH4_x2_y1_z1_2_materialName: str = "", WH4_x2_y1_z1_2_quantity: float = 0.0,
@@ -630,7 +630,12 @@ class BioyondCellWorkstation(BioyondWorkstation):
         response = self._post_lims("/api/lims/order/orders", orders)
         print(response)
         # 等待任务报送成功
-        order_code = response.get("data", {}).get("orderCode")
+        data_list = response.get("data", [])
+        if data_list:
+            order_code = data_list[0].get("orderCode")
+        else:
+            order_code = None
+
         if not order_code:
             logger.error("上料任务未返回有效 orderCode！")
             return response
@@ -963,6 +968,119 @@ class BioyondCellWorkstation(BioyondWorkstation):
             logger.error(f"✗ 执行失败: {e}")
             return {"success": False, "error": str(e)}
 
+def create_material(
+        self,
+        material_name: str,
+        type_id: str,
+        warehouse_name: str,
+        location_name_or_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """创建单个物料并可选入库。
+        Args:
+            material_name: 物料名称（会优先匹配配置模板）。
+            type_id: 物料类型 ID（若为空则尝试从配置推断）。
+            warehouse_name: 需要入库的仓库名称；若为空则仅创建不入库。
+            location_name_or_id: 具体库位名称（如 A01）或库位 UUID，由用户指定。
+        Returns:
+            包含创建结果、物料ID以及入库结果的字典。
+        """
+        material_name = (material_name or "").strip()
+        if not material_name:
+            raise ValueError("material_name 不能为空")
+        resolved_type_id = (type_id or "").strip()
+        # 优先从 SOLID_LIQUID_MAPPINGS 中获取模板数据
+        template = SOLID_LIQUID_MAPPINGS.get(material_name)
+        if not template:
+            raise ValueError(f"在配置中未找到物料 {material_name} 的模板，请检查 SOLID_LIQUID_MAPPINGS。")
+        material_data: Dict[str, Any]
+        material_data = deepcopy(template)
+        # 最终确保 typeId 为调用方传入的值
+        if resolved_type_id:
+            material_data["typeId"] = resolved_type_id
+        material_data["name"] = material_name
+        # 生成唯一编码
+        def _generate_code(prefix: str) -> str:
+            normalized = re.sub(r"\W+", "_", prefix)
+            normalized = normalized.strip("_") or "material"
+            return f"{normalized}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
+        if not material_data.get("code"):
+            material_data["code"] = _generate_code(material_name)
+        if not material_data.get("barCode"):
+            material_data["barCode"] = ""
+        # 处理数量字段类型
+        def _to_number(value: Any, default: float = 0.0) -> float:
+            try:
+                if value is None:
+                    return default
+                if isinstance(value, (int, float)):
+                    return float(value)
+                if isinstance(value, str) and value.strip() == "":
+                    return default
+                return float(value)
+            except (TypeError, ValueError):
+                return default
+        material_data["quantity"] = _to_number(material_data.get("quantity"), 1.0)
+        material_data["warningQuantity"] = _to_number(material_data.get("warningQuantity"), 0.0)
+        unit = material_data.get("unit") or "个"
+        material_data["unit"] = unit
+        if not material_data.get("parameters"):
+            material_data["parameters"] = json.dumps({"unit": unit}, ensure_ascii=False)
+        # 补充子物料信息
+        details = material_data.get("details") or []
+        if not isinstance(details, list):
+            logger.warning("details 字段不是列表，已忽略。")
+            details = []
+        else:
+            for idx, detail in enumerate(details, start=1):
+                if not isinstance(detail, dict):
+                    continue
+                if not detail.get("code"):
+                    detail["code"] = f"{material_data['code']}_{idx:02d}"
+                if not detail.get("name"):
+                    detail["name"] = f"{material_name}_detail_{idx:02d}"
+                if not detail.get("unit"):
+                    detail["unit"] = unit
+                if not detail.get("parameters"):
+                    detail["parameters"] = json.dumps({"unit": detail.get("unit", unit)}, ensure_ascii=False)
+                if "quantity" in detail:
+                    detail["quantity"] = _to_number(detail.get("quantity"), 1.0)
+        material_data["details"] = details
+        create_result = self._post_lims("/api/lims/storage/material", material_data)
+        # 解析创建结果中的物料 ID
+        material_id: Optional[str] = None
+        if isinstance(create_result, dict):
+            data_field = create_result.get("data")
+            if isinstance(data_field, str):
+                material_id = data_field
+            elif isinstance(data_field, dict):
+                material_id = data_field.get("id") or data_field.get("materialId")
+        inbound_result: Optional[Dict[str, Any]] = None
+        location_id: Optional[str] = None
+        # 按用户指定位置入库
+        if warehouse_name and material_id and location_name_or_id:
+            try:
+                location_ids, position_names = self._load_warehouse_locations(warehouse_name)
+                position_to_id = {name: loc_id for name, loc_id in zip(position_names, location_ids)}
+                target_location_id = position_to_id.get(location_name_or_id, location_name_or_id)
+                if target_location_id:
+                    location_id = target_location_id
+                    inbound_result = self.storage_inbound(material_id, target_location_id)
+                else:
+                    inbound_result = {"error": f"未找到匹配的库位: {location_name_or_id}"}
+            except Exception as exc:
+                logger.error(f"获取仓库 {warehouse_name} 位置失败: {exc}")
+                inbound_result = {"error": str(exc)}
+        return {
+            "success": bool(isinstance(create_result, dict) and create_result.get("code") == 1 and material_id),
+            "material_name": material_name,
+            "material_id": material_id,
+            "warehouse": warehouse_name,
+            "location_id": location_id,
+            "location_name_or_id": location_name_or_id,
+            "create_result": create_result,
+            "inbound_result": inbound_result,
+        }
+
 
 # --------------------------------
 
@@ -971,7 +1089,7 @@ if __name__ == "__main__":
     lab_registry.setup()
     ws = BioyondCellWorkstation()
     # logger.info(ws.scheduler_stop())
-    # logger.info(ws.scheduler_start())
+    logger.info(ws.scheduler_start())
     
     # results = ws.create_materials(SOLID_LIQUID_MAPPINGS)
     # for r in results:
@@ -980,11 +1098,11 @@ if __name__ == "__main__":
     # result = ws.create_and_inbound_materials()
     
     # 继续后续流程
-    # logger.info(ws.auto_feeding4to3()) #搬运物料到3号箱
+    logger.info(ws.auto_feeding4to3()) #搬运物料到3号箱
     # # 使用正斜杠或 Path 对象来指定文件路径
-    # excel_path = Path("unilabos\\devices\\workstation\\bioyond_studio\\bioyond_cell\\2025092701.xlsx")
-    # logger.info(ws.create_orders(excel_path))
-    # logger.info(ws.transfer_3_to_2_to_1())
+    excel_path = Path("unilabos\\devices\\workstation\\bioyond_studio\\bioyond_cell\\2025092701.xlsx")
+    logger.info(ws.create_orders(excel_path))
+    logger.info(ws.transfer_3_to_2_to_1())
 
     # logger.info(ws.transfer_1_to_2())
     # logger.info(ws.scheduler_start())

unilabos/devices/workstation/bioyond_studio/config.py

@@ -16,7 +16,7 @@ API_CONFIG = {
     "report_token": os.getenv("BIOYOND_REPORT_TOKEN", "CHANGE_ME_TOKEN"),
     
     # HTTP 服务配置
-    "HTTP_host": os.getenv("BIOYOND_HTTP_HOST", "172.21.32.210"),  # HTTP服务监听地址，监听计算机飞连ip地址
+    "HTTP_host": os.getenv("BIOYOND_HTTP_HOST", "172.21.33.174"),  # HTTP服务监听地址，监听计算机飞连ip地址
     "HTTP_port": int(os.getenv("BIOYOND_HTTP_PORT", "8080")),
     "debug_mode": False,# 调试模式
 }