HR物料同步，前端展示位置修复

[ZiWei09]

This pull request introduces significant improvements to the integration between the UniLabOS system and the Bioyond workstation, focusing on more robust material synchronization, enhanced error handling, and improved configuration mapping. The changes affect both the configuration files and the core synchronization logic, making the system more reliable and easier to debug.

Enhancements to Bioyond integration and material synchronization:

• Material Synchronization Logic:
  The sync_from_external and sync_to_external methods in station.py now handle both sample and reagent types by querying Bioyond for both typeMode=1 and typeMode=2, combining results for more complete material data synchronization. Logging has been improved for better traceability. [1] [2]

• Robust Material Inbound Workflow:
  The material inbound process now checks for location occupancy, verifies that the target location is available, and confirms successful storage by re-querying Bioyond. It also distinguishes between creating new materials and using existing ones, with detailed logging and error handling throughout the process.

Configuration and mapping improvements:

• Material Type Mapping Updates:
  The configuration files for both dispensing and reaction stations have been updated to use more explicit and consistent material type mappings, switching from Chinese names to standardized Bioyond identifiers, and including UUIDs for each type. [1] [2]

• Addition of New Experiment Configuration:
  A new experiment configuration file, ICCAS506.json, has been added, defining nodes and device/deck relationships for Bioyond dispensing and reaction stations, with detailed mappings and setup information.

Error handling and logging:

• Improved Exception Management:
  Exception handling in synchronization methods has been refined to ensure errors are logged clearly, and stack traces are printed only when necessary, facilitating easier debugging and maintenance. [1] [2]

Summary by Sourcery

Add station-specific bottle and carrier definitions, revise warehouse layout handling, and enhance Bioyond synchronization with improved error handling, logging, and configuration mapping.

New Features:

• Add dispensing and reaction station bottle and carrier factory functions (e.g., 8-stock, 6-vial, reagent and flask carriers, and tip boxes).
• Introduce configurable row-major/column-major warehouse_factory layouts and a new reagent stack warehouse.
• Implement backward-compatible wrappers for PolymerStation carriers.
• Extend station synchronization to query both sample (typeMode=1) and reagent (typeMode=2) materials, check location occupancy, and re-verify inbound storage.
• Add resource_tree_add and resource_tree_transfer methods to handle incremental and transfer synchronization to Bioyond.

Enhancements:

• Replace print statements with logger calls and enrich logging across sync methods and graphio conversions.
• Improve resource_bioyond_to_plr and resource_plr_to_bioyond mappings with reverse lookup, unique material naming, robust typeId validation, and fixed coordinate translations.
• Update bottle_carriers.yaml and bottles.yaml registries to include new resources and maintain backward compatibility.

Documentation:

• Add new experiment configuration file ICCAS506.json for Bioyond workstation integration.

[sourcery-ai]

Reviewer's Guide

This PR deeply refactors the Bioyond integration by extending synchronization logic to handle both sample and reagent types with a robust inbound workflow and unified configuration merging; expands dispensing/reaction station resource definitions with new carriers and containers plus registry entries; adds flexible warehouse ordering (row/col-major) and new reagent stacks; enriches PLR↔Bioyond mapping with reverse lookups, unique naming, and detail inference; standardizes logging and error handling across modules; and includes a new ICCAS506 experiment configuration.

Sequence diagram for improved material synchronization between UniLabOS and Bioyond

sequenceDiagram
    participant UniLabOS
    participant BioyondAPI
    UniLabOS->>BioyondAPI: Query stock_material(typeMode=1)
    BioyondAPI-->>UniLabOS: Return sample type materials
    UniLabOS->>BioyondAPI: Query stock_material(typeMode=2)
    BioyondAPI-->>UniLabOS: Return reagent type materials
    UniLabOS->>UniLabOS: Combine results
    UniLabOS->>UniLabOS: Convert to PLR format
    UniLabOS->>UniLabOS: Log results and errors

Loading

Sequence diagram for robust material inbound workflow

sequenceDiagram
    participant UniLabOS
    participant BioyondAPI
    UniLabOS->>BioyondAPI: Check if material exists
    alt Material exists
        UniLabOS->>BioyondAPI: Move material to target location
    else Material does not exist
        UniLabOS->>BioyondAPI: Add new material
        BioyondAPI-->>UniLabOS: Return material ID
    end
    UniLabOS->>BioyondAPI: Check if target location is occupied
    alt Location occupied by same material
        UniLabOS->>UniLabOS: Skip inbound
    else Location occupied by other material
        UniLabOS->>UniLabOS: Abort inbound
    else Location available
        UniLabOS->>BioyondAPI: Inbound material to location
        BioyondAPI-->>UniLabOS: Confirm inbound
        UniLabOS->>BioyondAPI: Re-query to verify location
    end

Loading

ER diagram for material type mapping and registry updates

erDiagram
    MATERIAL_TYPE_MAPPINGS {
      model string
      display_name string
      uuid string
    }
    BOTTLE_CARRIERS {
      name string
      model string
      registry_type string
    }
    BOTTLES {
      name string
      model string
      registry_type string
    }
    MATERIAL_TYPE_MAPPINGS ||--o{ BOTTLE_CARRIERS : maps
    MATERIAL_TYPE_MAPPINGS ||--o{ BOTTLES : maps
    BOTTLE_CARRIERS ||--o{ BOTTLES : contains

Loading

Class diagram for new and updated bottle carriers and containers

classDiagram
    class BottleCarrier {
      +name: str
      +size_x: float
      +size_y: float
      +size_z: float
      +sites: dict
      +model: str
      +num_items_x: int
      +num_items_y: int
      +num_items_z: int
    }
    class Bottle {
      +name: str
      +diameter: float
      +height: float
      +max_volume: float
      +barcode: str
      +model: str
    }
    class BIOYOND_DispensingStation_8StockCarrier {
      +BottleCarrier
    }
    class BIOYOND_DispensingStation_6VialCarrier {
      +BottleCarrier
    }
    class BIOYOND_DispensingStation_1BottleCarrier {
      +BottleCarrier
    }
    class BIOYOND_DispensingStation_1FlaskCarrier {
      +BottleCarrier
    }
    class BIOYOND_ReactionStation_6StockCarrier {
      +BottleCarrier
    }
    class BIOYOND_ReactionStation_1BottleCarrier {
      +BottleCarrier
    }
    class BIOYOND_ReactionStation_1FlaskCarrier {
      +BottleCarrier
    }
    class BIOYOND_DispensingStation_Solid_Stock {
      +Bottle
    }
    class BIOYOND_DispensingStation_Solid_Vial {
      +Bottle
    }
    class BIOYOND_DispensingStation_Liquid_Vial {
      +Bottle
    }
    class BIOYOND_DispensingStation_Reagent_Bottle {
      +Bottle
    }
    class BIOYOND_ReactionStation_Reactor {
      +Bottle
    }
    class BIOYOND_ReactionStation_Solid_Vial {
      +Bottle
    }
    class BIOYOND_ReactionStation_Liquid_Vial {
      +Bottle
    }
    class BIOYOND_ReactionStation_Reagent_Bottle {
      +Bottle
    }
    class BIOYOND_ReactionStation_Flask {
      +Bottle
    }
    BIOYOND_DispensingStation_8StockCarrier --|> BottleCarrier
    BIOYOND_DispensingStation_6VialCarrier --|> BottleCarrier
    BIOYOND_DispensingStation_1BottleCarrier --|> BottleCarrier
    BIOYOND_DispensingStation_1FlaskCarrier --|> BottleCarrier
    BIOYOND_ReactionStation_6StockCarrier --|> BottleCarrier
    BIOYOND_ReactionStation_1BottleCarrier --|> BottleCarrier
    BIOYOND_ReactionStation_1FlaskCarrier --|> BottleCarrier
    BIOYOND_DispensingStation_Solid_Stock --|> Bottle
    BIOYOND_DispensingStation_Solid_Vial --|> Bottle
    BIOYOND_DispensingStation_Liquid_Vial --|> Bottle
    BIOYOND_DispensingStation_Reagent_Bottle --|> Bottle
    BIOYOND_ReactionStation_Reactor --|> Bottle
    BIOYOND_ReactionStation_Solid_Vial --|> Bottle
    BIOYOND_ReactionStation_Liquid_Vial --|> Bottle
    BIOYOND_ReactionStation_Reagent_Bottle --|> Bottle
    BIOYOND_ReactionStation_Flask --|> Bottle

Loading

Class diagram for WareHouse with ordering_layout

classDiagram
    class WareHouse {
      +name: str
      +size_x: float
      +size_y: float
      +size_z: float
      +num_items_x: int
      +num_items_y: int
      +num_items_z: int
      +sites: dict
      +ordering_layout: str
      +serialize()
    }

Loading

File-Level Changes

Change	Details	Files
Overhaul of sync_from_external and sync_to_external workflows	Query both typeMode=1 and typeMode=2 and merge results Implement multi-step inbound in sync_to_external (ID reuse, occupancy check, re-query for verification) Merge default and custom Bioyond configs; refine exception handling and logging Iterate resources in resource_tree_add with ID check; hook resource_tree_transfer	unilabos/devices/workstation/bioyond_studio/station.py unilabos/ros/nodes/base_device_node.py
Extended bottle carrier definitions for dispensing and reaction stations	Add DispensingStation_* and ReactionStation_* carrier factory functions with correct sizes and ordering Rename and deprecate old PolymerStation_* wrappers for backward compatibility Adjust model strings and carrier layouts (2×4, 2×3, 1×1)	unilabos/resources/bioyond/bottle_carriers.py unilabos/registry/resources/bioyond/bottle_carriers.yaml
Added new container factories for dispensing/reaction stations	Implement Bottle and TipBox factories for DispensingStation and ReactionStation Define reactor, vial, flask, tip box models with barcode support Update registry entries for all new container types	unilabos/resources/bioyond/bottles.py unilabos/registry/resources/bioyond/bottles.yaml
Flexible warehouse ordering and new reagent stacks	Introduce layout parameter and ordering_layout attribute (row-major vs col-major) in warehouse_factory and WareHouse Add bioyond_warehouse_reagent_stack and update decks setup Serialize ordering_layout and adjust coordinate calculations	unilabos/resources/warehouse.py unilabos/resources/bioyond/warehouses.py unilabos/resources/bioyond/decks.py
Enhanced PLR↔Bioyond resource mapping in graphio	Build reverse TypeName→(model,UUID) mapping and log entries Assign unique suffixes to same-name materials Infer detail types for sample plates and validate indices Streamline resource_plr_to_bioyond typeId lookups with error reports	unilabos/resources/graphio.py
Unified logging and improved error handling	Replace print statements with logger.warning/debug Remove raw traceback prints or wrap them under controlled logging Add contextual log messages throughout sync and mapping code	unilabos/devices/workstation/bioyond_studio/station.py unilabos/resources/graphio.py
Addition of new ICCAS506 experiment configuration	Create ICCAS506.json defining nodes and device/deck relationships	test/experiments/ICCAS506.json

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey there - I've reviewed your changes - here's some feedback:

• The sync_to_external method is very large and deeply nested—consider breaking it into smaller helper functions (e.g. for location resolution, inbound checks, creation vs. move) to improve readability and maintainability.
• You’re querying stock_material twice in both sync methods; extract that logic (fetching typeMode 1 and 2 and merging results) into a shared helper to reduce duplication and ensure consistency.
• Instead of manual traceback.print_exc() and inline imports in exception blocks, use logger.exception() (or logger.error with exc_info) to capture stack traces consistently through your logging framework.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The sync_to_external method is very large and deeply nested—consider breaking it into smaller helper functions (e.g. for location resolution, inbound checks, creation vs. move) to improve readability and maintainability.
- You’re querying stock_material twice in both sync methods; extract that logic (fetching typeMode 1 and 2 and merging results) into a shared helper to reduce duplication and ensure consistency.
- Instead of manual traceback.print_exc() and inline imports in exception blocks, use logger.exception() (or logger.error with exc_info) to capture stack traces consistently through your logging framework.

## Individual Comments

### Comment 1
<location> `unilabos/resources/warehouse.py:27-30` </location>
<code_context>
     model: Optional[str] = None,
-    col_offset: int = 0,  # 新增：列起始偏移量，用于生成A05-D08等命名
+    col_offset: int = 0,  # 列起始偏移量，用于生成A05-D08等命名
+    layout: str = "col-major",  # 新增：排序方式，"col-major"=列优先，"row-major"=行优先
 ):
-    # 创建16个板架位 (4层 x 4位置)
</code_context>

<issue_to_address>
**suggestion:** Consider validating the 'layout' parameter to ensure only supported values are used.

Adding explicit validation for 'layout' will prevent unsupported values and reduce the risk of configuration errors.

```suggestion
    layout: str = "col-major",  # 新增：排序方式，"col-major"=列优先，"row-major"=行优先
):
    # 验证 layout 参数
    if layout not in ("col-major", "row-major"):
        raise ValueError(f"Unsupported layout: {layout}. Supported values are 'col-major' and 'row-major'.")
    # 创建位置坐标
    locations = []
```
</issue_to_address>

### Comment 2
<location> `unilabos/resources/graphio.py:632-635` </location>
<code_context>
     """
     plr_materials = []

+    # 创建反向映射: {显示名称: (model, UUID)} -> 用于从 Bioyond typeName 查找 model
+    # 如果 type_mapping 的 key 已经是显示名称,则直接使用;否则创建反向映射
+    reverse_type_mapping = {}
+    for key, value in type_mapping.items():
+        # value 可能是 tuple 或 list: (显示名称, UUID) 或 [显示名称, UUID]
+        display_name = value[0] if isinstance(value, (tuple, list)) and len(value) >= 1 else None
</code_context>

<issue_to_address>
**suggestion (bug_risk):** Reverse mapping logic assumes display names are unique, which may not always be true.

Currently, duplicate display names in type_mapping will cause only the first entry to be retained in the reverse mapping. Please add a check to detect duplicates and raise a warning or error if found.
</issue_to_address>

### Comment 3
<location> `unilabos/resources/graphio.py:875-877` </location>
<code_context>
+            # 单个瓶子(非载架)类型的资源
             bottle = resource[0] if resource.capacity > 0 else resource
+
+            # 根据 resource.model 从 type_mapping 获取正确的 typeId
+            type_info = type_mapping.get(resource.model)
+            if type_info:
+                type_id = type_info[1]
+            else:
</code_context>

<issue_to_address>
**issue (bug_risk):** Defaulting to a hardcoded typeId for unmapped models may lead to incorrect data.

Using a default typeId when a model is missing from type_mapping risks misclassification. It would be safer to raise an error or enforce explicit mappings for all resource types.
</issue_to_address>

### Comment 4
<location> `unilabos/devices/workstation/bioyond_studio/station.py:457-466` </location>
<code_context>
         """
-        self.resource_synchronizer.sync_to_external(resources)
+        logger.info(f"[resource_tree_add] 开始同步 {len(resources)} 个资源到 Bioyond 系统")
+        for resource in resources:
+            try:
+                # 🔍 检查资源是否已有 Bioyond ID (避免重复入库)
+                extra_info = getattr(resource, "unilabos_extra", {})
+                material_bioyond_id = extra_info.get("material_bioyond_id")
+
+                if material_bioyond_id:
+                    logger.info(f"⏭️ [resource_tree_add] 跳过资源 {resource.name}: 已有 Bioyond ID ({material_bioyond_id[:8]}...)，可能由 transfer 已处理")
+                    continue
+
+                logger.info(f"[resource_tree_add] 同步资源: {resource}")
+                self.resource_synchronizer.sync_to_external(resource)
+            except Exception as e:
+                logger.error(f"[resource_tree_add] 同步资源失败 {resource}: {e}")
</code_context>

<issue_to_address>
**issue (bug_risk):** Iterating over resources and skipping those with Bioyond ID may miss updates for moved resources.

Please ensure that resources with existing Bioyond IDs are checked for location changes and updated in Bioyond if necessary.
</issue_to_address>

### Comment 5
<location> `unilabos/ros/nodes/base_device_node.py:611-619` </location>
<code_context>
                         f"物料{plr_resource}请求挂载到{parent_resource}，额外参数：{additional_params}"
                     )
+
+                    # ⭐ assign 之前，需要从 resources 列表中移除
+                    # 因为资源将不再是顶级资源，而是成为 parent_resource 的子资源
+                    # 如果不移除，figure_resource 会找到两次：一次在 resources，一次在 parent 的 children
+                    resource_id = id(plr_resource)
+                    for i, r in enumerate(self.resource_tracker.resources):
+                        if id(r) == resource_id:
+                            self.resource_tracker.resources.pop(i)
+                            self.lab_logger().debug(f"从顶级资源列表中移除 {plr_resource.name}（即将成为 {parent_resource.name} 的子资源）")
+                            break
+
                     parent_resource.assign_child_resource(
</code_context>

<issue_to_address>
**suggestion:** Removing resources by id may not handle all references or edge cases.

Removing a resource from the top-level list may not update other references, such as those in trackers or caches, potentially causing inconsistencies. Please review all places where the resource might be referenced and ensure proper cleanup.

Suggested implementation:

```python
                    resource_id = id(plr_resource)
                    for i, r in enumerate(self.resource_tracker.resources):
                        if id(r) == resource_id:
                            self.resource_tracker.resources.pop(i)
                            self.lab_logger().debug(f"从顶级资源列表中移除 {plr_resource.name}（即将成为 {parent_resource.name} 的子资源）")
                            # 清理 resource_tracker 相关引用
                            if hasattr(self.resource_tracker, "resource_cache"):
                                # 假设 resource_cache 是以 id 为 key 的 dict
                                if resource_id in self.resource_tracker.resource_cache:
                                    del self.resource_tracker.resource_cache[resource_id]
                                    self.lab_logger().debug(f"从 resource_cache 中移除 {plr_resource.name}")
                            # 如果有其他 tracker/caches，按需清理
                            # TODO: 如有其他引用结构，请在此处补充清理逻辑
                            break

```

请检查 `resource_tracker` 是否还有其他引用资源的结构（如 set、list、dict 等），并在上述代码块中补充相应的清理逻辑。如果有自定义缓存或索引，请确保同步移除相关条目，以避免悬挂引用和数据不一致。
</issue_to_address>

### Comment 6
<location> `unilabos/devices/workstation/bioyond_studio/station.py:98` </location>
<code_context>
    def sync_to_external(self, resource: Any) -> bool:
        """将本地物料数据变更同步到Bioyond系统"""
        try:
            # ✅ 跳过仓库类型的资源 - 仓库是容器，不是物料
            resource_category = getattr(resource, "category", None)
            if resource_category == "warehouse":
                logger.debug(f"[同步→Bioyond] 跳过仓库类型资源: {resource.name} (仓库是容器，不需要同步为物料)")
                return True

            logger.info(f"[同步→Bioyond] 收到物料变更: {resource.name}")

            # 获取物料的 Bioyond ID
            extra_info = getattr(resource, "unilabos_extra", {})
            material_bioyond_id = extra_info.get("material_bioyond_id")

            # ⭐ 如果没有 Bioyond ID，尝试从 Bioyond 系统中按名称查询
            if not material_bioyond_id:
                logger.warning(f"[同步→Bioyond] 物料 {resource.name} 没有 Bioyond ID，尝试按名称查询...")
                try:
                    # 查询所有类型的物料：0=耗材, 1=样品, 2=试剂
                    import json
                    all_materials = []

                    for type_mode in [0, 1, 2]:
                        query_params = json.dumps({
                            "typeMode": type_mode,
                            "filter": "",   # 空字符串表示查询所有
                            "includeDetail": True
                        })
                        materials = self.bioyond_api_client.stock_material(query_params)
                        if materials:
                            all_materials.extend(materials)

                    logger.info(f"[同步→Bioyond] 查询到 {len(all_materials)} 个物料")

                    # 按名称匹配
                    for mat in all_materials:
                        if mat.get("name") == resource.name:
                            material_bioyond_id = mat.get("id")
                            mat_type = mat.get("typeName", "未知")
                            logger.info(f"✅ 找到物料 {resource.name} ({mat_type}) 的 Bioyond ID: {material_bioyond_id[:8]}...")
                            # 保存 ID 到资源对象
                            extra_info["material_bioyond_id"] = material_bioyond_id
                            setattr(resource, "unilabos_extra", extra_info)
                            break

                    if not material_bioyond_id:
                        logger.warning(f"⚠️ 在 Bioyond 系统中未找到名为 {resource.name} 的物料")
                        logger.info(f"[同步→Bioyond] 这是一个新物料，将创建并入库到 Bioyond 系统")
                        # 不返回，继续执行后续的创建+入库流程
                except Exception as e:
                    logger.error(f"查询 Bioyond 物料失败: {e}")
                    return False

            # 检查是否有位置更新请求
            update_site = extra_info.get("update_resource_site")

            if not update_site:
                logger.debug(f"[同步→Bioyond] 物料 {resource.name} 无位置更新请求，跳过同步")
                return True

            # ===== 物料移动/创建流程 =====
            logger.info(f"[同步→Bioyond] 📍 物料 {resource.name} 目标库位: {update_site}")

            if material_bioyond_id:
                logger.info(f"[同步→Bioyond] 🔄 物料已存在于 Bioyond (ID: {material_bioyond_id[:8]}...)，执行移动操作")
            else:
                logger.info(f"[同步→Bioyond] ➕ 物料不存在于 Bioyond，将创建新物料并入库")

            # 第1步：获取仓库配置
            from .config import WAREHOUSE_MAPPING
            warehouse_mapping = WAREHOUSE_MAPPING

            # 确定目标仓库名称（优先使用 resource.parent.name）
            parent_name = None
            target_location_uuid = None

            # 如果资源有父节点，优先使用父节点名称
            if resource.parent is not None:
                parent_name = resource.parent.name
                logger.info(f"[同步→Bioyond] 从资源父节点获取仓库名称: {parent_name}")

                # 检查该仓库是否在配置中
                if parent_name in warehouse_mapping:
                    site_uuids = warehouse_mapping[parent_name].get("site_uuids", {})
                    if update_site in site_uuids:
                        target_location_uuid = site_uuids[update_site]
                        logger.info(f"[同步→Bioyond] 目标仓库: {parent_name}/{update_site}")
                        logger.info(f"[同步→Bioyond] 目标库位UUID: {target_location_uuid[:8]}...")
                    else:
                        logger.warning(f"⚠️ [同步→Bioyond] 仓库 {parent_name} 中没有库位 {update_site}")
                else:
                    logger.warning(f"⚠️ [同步→Bioyond] 仓库 {parent_name} 未在 WAREHOUSE_MAPPING 中配置")
                    parent_name = None

            # 如果没有找到，则遍历所有仓库查找
            if not parent_name or not target_location_uuid:
                logger.info(f"[同步→Bioyond] 从所有仓库中查找库位 {update_site}...")
                for warehouse_name, warehouse_info in warehouse_mapping.items():
                    site_uuids = warehouse_info.get("site_uuids", {})
                    if update_site in site_uuids:
                        parent_name = warehouse_name
                        target_location_uuid = site_uuids[update_site]
                        logger.info(f"[同步→Bioyond] 目标仓库: {parent_name}/{update_site}")
                        logger.info(f"[同步→Bioyond] 目标库位UUID: {target_location_uuid[:8]}...")
                        break

            if not parent_name or not target_location_uuid:
                logger.error(f"❌ [同步→Bioyond] 库位 {update_site} 没有在 WAREHOUSE_MAPPING 中配置")
                logger.debug(f"[同步→Bioyond] 可用仓库: {list(warehouse_mapping.keys())}")
                return False

            # 第2步：转换为 Bioyond 格式
            logger.info(f"[同步→Bioyond] 🔄 转换物料为 Bioyond 格式...")
            bioyond_material = resource_plr_to_bioyond(
                [resource],
                type_mapping=self.workstation.bioyond_config["material_type_mappings"],
                warehouse_mapping=self.workstation.bioyond_config["warehouse_mapping"]
            )[0]

            logger.debug(f"[同步→Bioyond] Bioyond 物料数据: {bioyond_material}")

            location_info = bioyond_material.pop("locations", None)
            logger.debug(f"[同步→Bioyond] 库位信息: {location_info}, 类型: {type(location_info)}")

            # 第3步：根据是否已有 Bioyond ID 决定创建还是使用现有物料
            if material_bioyond_id:
                # 物料已存在,直接使用现有 ID
                material_id = material_bioyond_id
                logger.info(f"✅ [同步→Bioyond] 使用已有物料 ID: {material_id[:8]}...")
            else:
                # 物料不存在,调用 API 创建新物料
                logger.info(f"[同步→Bioyond] 📤 调用 Bioyond API 添加物料...")
                material_id = self.bioyond_api_client.add_material(bioyond_material)

                if not material_id:
                    logger.error(f"❌ [同步→Bioyond] 添加物料失败，API 返回空")
                    return False

                logger.info(f"✅ [同步→Bioyond] 物料添加成功，Bioyond ID: {material_id[:8]}...")

                # 保存新创建的物料 ID 到资源对象
                extra_info["material_bioyond_id"] = material_id
                setattr(resource, "unilabos_extra", extra_info)

            # 第4步：物料入库前先检查目标库位是否被占用
            if location_info:
                logger.info(f"[同步→Bioyond] 📥 准备入库到库位 {update_site}...")

                # 处理不同的 location_info 数据结构
                if isinstance(location_info, list) and len(location_info) > 0:
                    location_id = location_info[0]["id"]
                elif isinstance(location_info, dict):
                    location_id = location_info["id"]
                else:
                    logger.warning(f"⚠️ [同步→Bioyond] 无效的库位信息格式: {location_info}")
                    location_id = None

                if location_id:
                    # 查询目标库位是否已有物料
                    logger.info(f"[同步→Bioyond] 🔍 检查库位 {update_site} (UUID: {location_id[:8]}...) 是否被占用...")

                    # 查询所有物料，检查是否有物料在目标库位
                    try:
                        all_materials_type1 = self.bioyond_api_client.stock_material('{"typeMode": 1, "includeDetail": true}')
                        all_materials_type2 = self.bioyond_api_client.stock_material('{"typeMode": 2, "includeDetail": true}')
                        all_materials = (all_materials_type1 or []) + (all_materials_type2 or [])

                        # 检查是否有物料已经在目标库位
                        location_occupied = False
                        occupying_material = None

                        for material in all_materials:
                            locations = material.get("locations", [])
                            for loc in locations:
                                if loc.get("id") == location_id:
                                    location_occupied = True
                                    occupying_material = material
                                    logger.warning(f"⚠️ [同步→Bioyond] 库位 {update_site} 已被占用！")
                                    logger.warning(f"   占用物料: {material.get('name')} (ID: {material.get('id', '')[:8]}...)")
                                    logger.warning(f"   占用位置: code={loc.get('code')}, x={loc.get('x')}, y={loc.get('y')}")
                                    break
                            if location_occupied:
                                break

                        if location_occupied:
                            # 如果是同一个物料（名称相同），说明已经入库过了，跳过
                            if occupying_material and occupying_material.get("name") == resource.name:
                                logger.info(f"✅ [同步→Bioyond] 物料 {resource.name} 已经在库位 {update_site}，跳过重复入库")
                                return True
                            else:
                                logger.error(f"❌ [同步→Bioyond] 库位 {update_site} 已被其他物料占用，拒绝入库")
                                return False

                        logger.info(f"✅ [同步→Bioyond] 库位 {update_site} 可用，准备入库...")

                    except Exception as e:
                        logger.warning(f"⚠️ [同步→Bioyond] 检查库位状态时发生异常: {e}，继续尝试入库...")

                    # 执行入库
                    logger.info(f"[同步→Bioyond] 📥 调用 Bioyond API 物料入库...")
                    response = self.bioyond_api_client.material_inbound(material_id, location_id)

                    # 注意：Bioyond API 成功时返回空字典 {}，所以不能用 if not response 判断
                    # 只要没有抛出异常，就认为成功（response 是 dict 类型，即使是 {} 也不是 None）
                    if response is not None:
                        logger.info(f"✅ [同步→Bioyond] 物料 {resource.name} 成功入库到 {update_site}")

                        # 入库成功后，重新查询验证物料实际入库位置
                        logger.info(f"[同步→Bioyond] 🔍 验证物料实际入库位置...")
                        try:
                            all_materials_type1 = self.bioyond_api_client.stock_material('{"typeMode": 1, "includeDetail": true}')
                            all_materials_type2 = self.bioyond_api_client.stock_material('{"typeMode": 2, "includeDetail": true}')
                            all_materials = (all_materials_type1 or []) + (all_materials_type2 or [])

                            for material in all_materials:
                                if material.get("id") == material_id:
                                    locations = material.get("locations", [])
                                    if locations:
                                        actual_loc = locations[0]
                                        logger.info(f"📍 [同步→Bioyond] 物料实际位置: code={actual_loc.get('code')}, "
                                                  f"warehouse={actual_loc.get('whName')}, "
                                                  f"x={actual_loc.get('x')}, y={actual_loc.get('y')}")

                                        # 验证 UUID 是否匹配
                                        if actual_loc.get("id") != location_id:
                                            logger.error(f"❌ [同步→Bioyond] UUID 不匹配！")
                                            logger.error(f"   预期 UUID: {location_id}")
                                            logger.error(f"   实际 UUID: {actual_loc.get('id')}")
                                            logger.error(f"   这说明配置文件中的 UUID 映射有误，请检查 config.py 中的 WAREHOUSE_MAPPING")
                                    break
                        except Exception as e:
                            logger.warning(f"⚠️ [同步→Bioyond] 验证入库位置时发生异常: {e}")
                    else:
                        logger.error(f"❌ [同步→Bioyond] 物料入库失败")
                        return False
                else:
                    logger.warning(f"⚠️ [同步→Bioyond] 无法获取库位 ID，跳过入库操作")
            else:
                logger.warning(f"⚠️ [同步→Bioyond] 物料没有库位信息，跳过入库操作")
            return True

        except Exception as e:
            logger.error(f"❌ [同步→Bioyond] 同步物料 {resource.name} 时发生异常: {e}")
            import traceback
            traceback.print_exc()
            return False

</code_context>

<issue_to_address>
**issue (code-quality):** We've found these issues:

- Use named expression to simplify assignment and conditional [×2] ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- Replace f-string with no interpolated values with string ([`remove-redundant-fstring`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-fstring/))
- Swap positions of nested conditionals ([`swap-nested-ifs`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/swap-nested-ifs/))
- Hoist nested repeated code outside conditional statements ([`hoist-similar-statement-from-if`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/hoist-similar-statement-from-if/))
- Low code quality found in BioyondResourceSynchronizer.sync\_to\_external - 4% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))

<br/><details><summary>Explanation</summary>




The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.

How can you solve this?

It might be worth refactoring this function to make it shorter and more readable.

- Reduce the function length by extracting pieces of functionality out into
  their own functions. This is the most important thing you can do - ideally a
  function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
  sits together within the function rather than being scattered.</details>
</issue_to_address>

### Comment 7
<location> `unilabos/devices/workstation/bioyond_studio/station.py:461-463` </location>
<code_context>
    def resource_tree_add(self, resources: List[ResourcePLR]) -> None:
        """添加资源到资源树并更新ROS节点

        Args:
            resources (List[ResourcePLR]): 要添加的资源列表
        """
        logger.info(f"[resource_tree_add] 开始同步 {len(resources)} 个资源到 Bioyond 系统")
        for resource in resources:
            try:
                # 🔍 检查资源是否已有 Bioyond ID (避免重复入库)
                extra_info = getattr(resource, "unilabos_extra", {})
                material_bioyond_id = extra_info.get("material_bioyond_id")

                if material_bioyond_id:
                    logger.info(f"⏭️ [resource_tree_add] 跳过资源 {resource.name}: 已有 Bioyond ID ({material_bioyond_id[:8]}...)，可能由 transfer 已处理")
                    continue

                logger.info(f"[resource_tree_add] 同步资源: {resource}")
                self.resource_synchronizer.sync_to_external(resource)
            except Exception as e:
                logger.error(f"[resource_tree_add] 同步资源失败 {resource}: {e}")
                import traceback
                traceback.print_exc()

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))

```suggestion
                if material_bioyond_id := extra_info.get("material_bioyond_id"):
```
</issue_to_address>

### Comment 8
<location> `unilabos/devices/workstation/bioyond_studio/station.py:491-493` </location>
<code_context>
    def resource_tree_transfer(self, old_parent: Optional[ResourcePLR], resource: ResourcePLR, new_parent: ResourcePLR) -> None:
        """处理资源在设备间迁移时的同步

        当资源从一个设备迁移到 BioyondWorkstation 时，需要同步到 Bioyond 系统

        Args:
            old_parent: 资源的原父节点（可能为 None）
            resource: 要迁移的资源
            new_parent: 资源的新父节点
        """
        logger.info(f"[resource_tree_transfer] 资源迁移: {resource.name}")
        logger.info(f"  旧父节点: {old_parent.name if old_parent else 'None'}")
        logger.info(f"  新父节点: {new_parent.name}")

        try:
            # 同步资源到 Bioyond 系统
            logger.info(f"[resource_tree_transfer] 开始同步资源 {resource.name} 到 Bioyond 系统")
            result = self.resource_synchronizer.sync_to_external(resource)

            if result:
                logger.info(f"✅ [resource_tree_transfer] 资源 {resource.name} 成功同步到 Bioyond 系统")
            else:
                logger.warning(f"⚠️ [resource_tree_transfer] 资源 {resource.name} 同步到 Bioyond 系统失败")

        except Exception as e:
            logger.error(f"❌ [resource_tree_transfer] 资源 {resource.name} 同步异常: {e}")
            import traceback
            traceback.print_exc()

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))

```suggestion
            if result := self.resource_synchronizer.sync_to_external(resource):
```
</issue_to_address>

### Comment 9
<location> `unilabos/resources/graphio.py:588-590` </location>
<code_context>
def resource_plr_to_ulab(resource_plr: "ResourcePLR", parent_name: str = None, with_children=True):
    def replace_plr_type_to_ulab(source: str):
        replace_info = {
            "plate": "plate",
            "well": "well",
            "tip_spot": "tip_spot",
            "trash": "trash",
            "deck": "deck",
            "tip_rack": "tip_rack",
            "warehouse": "warehouse",
            "container": "container",
        }
        if source in replace_info:
            return replace_info[source]
        else:
            logger.warning(f"转换pylabrobot的时候，出现未知类型: {source}")
            return source

    def resource_plr_to_ulab_inner(d: dict, all_states: dict, child=True) -> dict:
        r = {
            "id": d["name"],
            "name": d["name"],
            "sample_id": None,
            "children": [resource_plr_to_ulab_inner(child, all_states) for child in d["children"]] if child else [],
            "parent": d["parent_name"] if d["parent_name"] else parent_name if parent_name else None,
            "type": replace_plr_type_to_ulab(d.get("category")),  # FIXME plr自带的type是python class name
            "class": d.get("class", ""),
            "position": (
                {"x": d["location"]["x"], "y": d["location"]["y"], "z": d["location"]["z"]}
                if d["location"]
                else {"x": 0, "y": 0, "z": 0}
            ),
            "config": {k: v for k, v in d.items() if k not in ["name", "children", "parent_name", "location"]},
            "data": all_states[d["name"]],
        }
        return r

    d = resource_plr.serialize()
    all_states = resource_plr.serialize_all_state()
    r = resource_plr_to_ulab_inner(d, all_states, with_children)

    return r

</code_context>

<issue_to_address>
**suggestion (code-quality):** Remove unnecessary else after guard condition ([`remove-unnecessary-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-unnecessary-else/))

```suggestion
        logger.warning(f"转换pylabrobot的时候，出现未知类型: {source}")
        return source
```
</issue_to_address>

### Comment 10
<location> `unilabos/resources/graphio.py:829` </location>
<code_context>
def resource_plr_to_bioyond(plr_resources: list[ResourcePLR], type_mapping: dict = {}, warehouse_mapping: dict = {}) -> list[dict]:
    bioyond_materials = []
    for resource in plr_resources:
        if isinstance(resource, BottleCarrier):
            # 获取 BottleCarrier 的类型映射
            type_info = type_mapping.get(resource.model)
            if not type_info:
                logger.error(f"❌ [PLR→Bioyond] BottleCarrier 资源 '{resource.name}' 的 model '{resource.model}' 不在 type_mapping 中")
                logger.debug(f"[PLR→Bioyond] 可用的 type_mapping 键: {list(type_mapping.keys())}")
                raise ValueError(f"资源 model '{resource.model}' 未在 MATERIAL_TYPE_MAPPINGS 中配置")

            material = {
                "typeId": type_info[1],
                "name": resource.name,
                "unit": "个",
                "quantity": 1,
                "details": [],
                "Parameters": "{}"
            }
            for bottle in resource.children:
                if isinstance(resource, ItemizedCarrier):
                    site = resource.get_child_identifier(bottle)
                else:
                    site = {"x": bottle.location.x - 1, "y": bottle.location.y - 1}

                # 获取子物料的类型映射
                bottle_type_info = type_mapping.get(bottle.model)
                if not bottle_type_info:
                    logger.error(f"❌ [PLR→Bioyond] 子物料 '{bottle.name}' 的 model '{bottle.model}' 不在 type_mapping 中")
                    raise ValueError(f"子物料 model '{bottle.model}' 未在 MATERIAL_TYPE_MAPPINGS 中配置")

                detail_item = {
                    "typeId": bottle_type_info[1],
                    "name": bottle.name,
                    "code": bottle.code if hasattr(bottle, "code") else "",
                    "quantity": sum(qty for _, qty in bottle.tracker.liquids) if hasattr(bottle, "tracker") else 0,
                    "x": site["x"] + 1,
                    "y": site["y"] + 1,
                    "molecular": 1,
                    "Parameters": json.dumps({"molecular": 1})
                }
                material["details"].append(detail_item)
        else:
            # 单个瓶子(非载架)类型的资源
            bottle = resource[0] if resource.capacity > 0 else resource

            # 根据 resource.model 从 type_mapping 获取正确的 typeId
            type_info = type_mapping.get(resource.model)
            if type_info:
                type_id = type_info[1]
            else:
                # 如果找不到映射，记录警告并使用默认值
                logger.warning(f"[PLR→Bioyond] 资源 {resource.name} 的 model '{resource.model}' 不在 type_mapping 中，使用默认烧杯类型")
                type_id = "3a14196b-24f2-ca49-9081-0cab8021bf1a"  # 默认使用烧杯类型

            material = {
                "typeId": type_id,
                "name": resource.name if hasattr(resource, "name") else "",
                "unit": "个",  # 修复：Bioyond API 要求 unit 字段不能为空
                "quantity": sum(qty for _, qty in bottle.tracker.liquids) if hasattr(bottle, "tracker") else 0,
                "Parameters": "{}"
            }

        if resource.parent is not None and isinstance(resource.parent, ItemizedCarrier):
            site_in_parent = resource.parent.get_child_identifier(resource)

            material["locations"] = [
                {
                    "id": warehouse_mapping[resource.parent.name]["site_uuids"][site_in_parent["identifier"]],
                    "whid": warehouse_mapping[resource.parent.name]["uuid"],
                    "whName": resource.parent.name,
                    "x": site_in_parent["x"] + 1,
                    "y": site_in_parent["y"] + 1,
                    "z": 1,
                    "quantity": 0
                }
            ]

        bioyond_materials.append(material)
    return bioyond_materials

</code_context>

<issue_to_address>
**issue (code-quality):** We've found these issues:

- Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- Replace mutable default arguments with None ([`default-mutable-arg`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/default-mutable-arg/))
- Low code quality found in resource\_plr\_to\_bioyond - 13% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))

<br/><details><summary>Explanation</summary>

The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.

How can you solve this?

It might be worth refactoring this function to make it shorter and more readable.

- Reduce the function length by extracting pieces of functionality out into
  their own functions. This is the most important thing you can do - ideally a
  function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
  sits together within the function rather than being scattered.</details>
</issue_to_address>

### Comment 11
<location> `unilabos/resources/warehouse.py:69` </location>
<code_context>
def warehouse_factory(
    name: str,
    num_items_x: int = 1,
    num_items_y: int = 4,
    num_items_z: int = 4,
    dx: float = 137.0,
    dy: float = 96.0,
    dz: float = 120.0,
    item_dx: float = 10.0,
    item_dy: float = 10.0,
    item_dz: float = 10.0,
    removed_positions: Optional[List[int]] = None,
    empty: bool = False,
    category: str = "warehouse",
    model: Optional[str] = None,
    col_offset: int = 0,  # 列起始偏移量，用于生成A05-D08等命名
    layout: str = "col-major",  # 新增：排序方式，"col-major"=列优先，"row-major"=行优先
):
    # 创建位置坐标
    locations = []

    for layer in range(num_items_z):  # 层
        for row in range(num_items_y):  # 行
            for col in range(num_items_x):  # 列
                # 计算位置
                x = dx + col * item_dx

                # 根据 layout 决定 y 坐标计算
                if layout == "row-major":
                    # 行优先：row=0(A行) 应该显示在上方，需要较小的 y 值
                    y = dy + row * item_dy
                else:
                    # 列优先：保持原逻辑（row=0 对应较大的 y）
                    y = dy + (num_items_y - row - 1) * item_dy

                z = dz + (num_items_z - layer - 1) * item_dz
                locations.append(Coordinate(x, y, z))
    if removed_positions:
        locations = [loc for i, loc in enumerate(locations) if i not in removed_positions]
    _sites = create_homogeneous_resources(
        klass=ResourceHolder,
        locations=locations,
        resource_size_x=127.0,
        resource_size_y=86.0,
        name_prefix=name,
    )
    len_x, len_y = (num_items_x, num_items_y) if num_items_z == 1 else (num_items_y, num_items_z) if num_items_x == 1 else (num_items_x, num_items_z)

    # 根据 layout 参数生成不同的排序方式
    # 注意：物理位置的 y 坐标是倒序的 (row=0 时 y 最大，对应前端显示的顶部)
    if layout == "row-major":
        # 行优先顺序: A01,A02,A03,A04, B01,B02,B03,B04
        # locations[0] 对应 row=0, y最大（前端顶部）→ 应该是 A01
        keys = [f"{LETTERS[j]}{i + 1 + col_offset:02d}" for j in range(len_y) for i in range(len_x)]
    else:
        # 列优先顺序: A01,B01,C01,D01, A02,B02,C02,D02
        keys = [f"{LETTERS[j]}{i + 1 + col_offset:02d}" for i in range(len_x) for j in range(len_y)]

    sites = {i: site for i, site in zip(keys, _sites.values())}

    return WareHouse(
        name=name,
        size_x=dx + item_dx * num_items_x,
        size_y=dy + item_dy * num_items_y,
        size_z=dz + item_dz * num_items_z,
        num_items_x = num_items_x,
        num_items_y = num_items_y,
        num_items_z = num_items_z,
        ordering_layout=layout,  # 传递排序方式到 ordering_layout
        # ordered_items=ordered_items,
        # ordering=ordering,
        sites=sites,
        category=category,
        model=model,
    )

</code_context>

<issue_to_address>
**suggestion (code-quality):** We've found these issues:

- Replace identity comprehension with call to collection constructor ([`identity-comprehension`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/identity-comprehension/))
- Low code quality found in warehouse\_factory - 23% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))

```suggestion
    sites = dict(zip(keys, _sites.values()))
```

<br/><details><summary>Explanation</summary>Convert list/set/tuple comprehensions that do not change the input elements into.

#### Before

```python
# List comprehensions
[item for item in coll]
[item for item in friends.names()]

# Dict comprehensions
{k: v for k, v in coll}
{k: v for k, v in coll.items()}  # Only if we know coll is a `dict`

# Unneeded call to `.items()`
dict(coll.items())  # Only if we know coll is a `dict`

# Set comprehensions
{item for item in coll}
```

#### After

```python
# List comprehensions
list(iter(coll))
list(iter(friends.names()))

# Dict comprehensions
dict(coll)
dict(coll)

# Unneeded call to `.items()`
dict(coll)

# Set comprehensions
set(coll)
```

All these comprehensions are just creating a copy of the original collection.
They can all be simplified by simply constructing a new collection directly. The
resulting code is easier to read and shows the intent more clearly.
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.

How can you solve this?

It might be worth refactoring this function to make it shorter and more readable.

- Reduce the function length by extracting pieces of functionality out into
  their own functions. This is the most important thing you can do - ideally a
  function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
  sits together within the function rather than being scattered.</details>
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

suggestion: Consider validating the 'layout' parameter to ensure only supported values are used.

Adding explicit validation for 'layout' will prevent unsupported values and reduce the risk of configuration errors.

Suggested change

	layout: str = "col-major", # 新增：排序方式，"col-major"=列优先，"row-major"=行优先
	):
	# 创建16个板架位 (4层 x 4位置)
	# 创建位置坐标
	locations = []
	layout: str = "col-major", # 新增：排序方式，"col-major"=列优先，"row-major"=行优先
	):
	# 验证 layout 参数
	if layout not in ("col-major", "row-major"):
	raise ValueError(f"Unsupported layout: {layout}. Supported values are 'col-major' and 'row-major'.")
	# 创建位置坐标
	locations = []

[sourcery-ai]

suggestion (bug_risk): Reverse mapping logic assumes display names are unique, which may not always be true.

Currently, duplicate display names in type_mapping will cause only the first entry to be retained in the reverse mapping. Please add a check to detect duplicates and raise a warning or error if found.

[sourcery-ai]

issue (bug_risk): Defaulting to a hardcoded typeId for unmapped models may lead to incorrect data.

Using a default typeId when a model is missing from type_mapping risks misclassification. It would be safer to raise an error or enforce explicit mappings for all resource types.

[sourcery-ai]

suggestion (code-quality): Remove unnecessary else after guard condition (remove-unnecessary-else)

Suggested change

	else:
	print("转换pylabrobot的时候，出现未知类型", source)
	logger.warning(f"转换pylabrobot的时候，出现未知类型: {source}")
	return source
	logger.warning(f"转换pylabrobot的时候，出现未知类型: {source}")
	return source

[sourcery-ai]

suggestion (code-quality): We've found these issues:

• Replace identity comprehension with call to collection constructor (identity-comprehension)
• Low code quality found in warehouse_factory - 23% (low-code-quality)

Suggested change

	sites = {i: site for i, site in zip(keys, _sites.values())}
	sites = dict(zip(keys, _sites.values()))

Explanation

Convert list/set/tuple comprehensions that do not change the input elements into.

Before

# List comprehensions
[item for item in coll]
[item for item in friends.names()]

# Dict comprehensions
{k: v for k, v in coll}
{k: v for k, v in coll.items()}  # Only if we know coll is a `dict`

# Unneeded call to `.items()`
dict(coll.items())  # Only if we know coll is a `dict`

# Set comprehensions
{item for item in coll}

After

# List comprehensions
list(iter(coll))
list(iter(friends.names()))

# Dict comprehensions
dict(coll)
dict(coll)

# Unneeded call to `.items()`
dict(coll)

# Set comprehensions
set(coll)

All these comprehensions are just creating a copy of the original collection.
They can all be simplified by simply constructing a new collection directly. The
resulting code is easier to read and shows the intent more clearly.
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.

How can you solve this?

It might be worth refactoring this function to make it shorter and more readable.

• Reduce the function length by extracting pieces of functionality out into
  their own functions. This is the most important thing you can do - ideally a
  function should be less than 10 lines.
• Reduce nesting, perhaps by introducing guard clauses to return early.
• Ensure that variables are tightly scoped, so that code using related concepts
  sits together within the function rather than being scattered.