Skip to content

Conversation

@ZiWei09
Copy link
Collaborator

@ZiWei09 ZiWei09 commented Nov 20, 2025

Summary by Sourcery

Introduce a unit-aware experiment design workflow by adding compute_experiment_design and its internal logic, enabling flexible ratio parsing, compound metadata lookup, and automated feeding order generation.

New Features:

  • Add compute_experiment_design method to calculate experiment parameters and feeding orders
  • Support simplified ratio string format parsing alongside JSON and dict inputs

Enhancements:

  • Integrate pint UnitRegistry and compound_info definitions for unit-aware calculations
  • Implement internal _generate_experiment_design to compute solution volumes, solvent requirements, and feeding sequence
  • Extend device registry YAML to register compute_experiment_design command with full schema and I/O mapping

@sourcery-ai
Copy link

sourcery-ai bot commented Nov 20, 2025

Reviewer's Guide

This PR introduces a new experiment design feature in the dispensing_station by integrating a pint-based unit system and compound metadata, implementing robust ratio parsing (simplified string and JSON), encapsulating the core calculation in a private helper, and exposing the functionality via an updated UniLab command in the device registry.

Sequence diagram for compute_experiment_design command flow

sequenceDiagram
participant Registry as "Device Registry"
participant Station as "DispensingStation"
participant Logger as "hardware_interface._logger"

Registry->>Station: Call compute_experiment_design(ratio, ...)
Station->>Logger: Log input ratio
alt Ratio is string
    Station->>Station: Parse ratio (simplified or JSON)
    Station->>Logger: Log parsed ratio
end
Station->>Station: Validate ratio and parameters
Station->>Station: Call _generate_experiment_design(...)
Station->>Logger: Log compound info
Station->>Station: Calculate solutions, titration, solvents, feeding_order
Station-->>Registry: Return result (solutions, titration, solvents, feeding_order, return_info)
Loading

ER diagram for compute_experiment_design result structure

erDiagram
    RESULT {
      string return_info
    }
    SOLUTION {
      string name
      int order
      float solid_mass
      float solvent_volume
      float concentration
      float volume_needed
      float molar_ratio
    }
    SOLID {
      string name
      int order
      float mass
      float molar_ratio
    }
    TITRATION {
      string name
      float main_portion
      float titration_portion
      float titration_solvent
    }
    SOLVENTS {
      float additional_solvent
      float total_liquid_volume
    }
    FEEDING_ORDER {
      int step
      string type
      string name
      float amount
      int order
      float titration_solvent
    }
    RESULT ||--o| SOLUTION : contains
    RESULT ||--o| SOLID : contains
    RESULT ||--o| TITRATION : contains
    RESULT ||--o| SOLVENTS : contains
    RESULT ||--o| FEEDING_ORDER : contains
Loading

Class diagram for updated DispensingStation with experiment design

classDiagram
class DispensingStation {
    - order_completion_status: dict
    - ureg: UnitRegistry
    - compound_info: dict
    + compute_experiment_design(ratio, wt_percent, m_tot, titration_percent): dict
    - _generate_experiment_design(ratio, wt_percent, m_tot, titration_percent): dict
}
DispensingStation --> UnitRegistry
DispensingStation --> BioyondException
Loading

File-Level Changes

Change Details Files
Initialize unit registry and define compound metadata
  • Instantiate pint.UnitRegistry in init
  • Populate compound_info with MolWt and FuncGroup mappings
unilabos/devices/workstation/bioyond_studio/dispensing_station.py
Implement compute_experiment_design with parsing and validation
  • Support simplified "MDA:0.5,PAPP:0.5" format and JSON strings
  • Convert wt_percent, m_tot, titration_percent to floats
  • Log parsing steps and raise meaningful BioyondException errors
  • Wrap call to internal helper and format returned dict
unilabos/devices/workstation/bioyond_studio/dispensing_station.py
Extract detailed calculation into _generate_experiment_design
  • Use pint units for density, solubility, mass and volume calculations
  • Separate amine and anhydride compounds, compute diamine solutions and solid masses
  • Calculate titration splits, additional solvents, scaling for minimum volume
  • Assemble ordered feeding steps and return comprehensive result dict
unilabos/devices/workstation/bioyond_studio/dispensing_station.py
Add compute_experiment_design command in device registry YAML
  • Define goal, default values, handlers, result mapping and JSON schema
  • Expose solutions, titration, solvents, feeding_order for downstream nodes
unilabos/registry/devices/bioyond_dispensing_station.yaml

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

Copy link

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

  • The compute_experiment_design method and _generate_experiment_design helper are very long—consider extracting parsing, calculation, and feeding-order logic into smaller, focused methods to improve readability and maintainability.
  • The compound_info dictionary in init could be relocated to a module-level constant or external config to keep the class constructor lean and make updates easier.
  • In _generate_experiment_design you mix ValueError and BioyondException—use a consistent exception type to standardize error handling across the compute_experiment_design workflow.
Prompt for AI Agents
Please address the comments from this code review:

## Overall Comments
- The compute_experiment_design method and _generate_experiment_design helper are very long—consider extracting parsing, calculation, and feeding-order logic into smaller, focused methods to improve readability and maintainability.
- The compound_info dictionary in __init__ could be relocated to a module-level constant or external config to keep the class constructor lean and make updates easier.
- In _generate_experiment_design you mix ValueError and BioyondException—use a consistent exception type to standardize error handling across the compute_experiment_design workflow.

## Individual Comments

### Comment 1
<location> `unilabos/devices/workstation/bioyond_studio/dispensing_station.py:286-289` </location>
<code_context>
+        compound_ratios = list(ratio.values())
+
+        # 验证所有化合物是否在 compound_info 中定义
+        undefined_compounds = [name for name in compound_names if name not in self.compound_info["MolWt"]]
+        if undefined_compounds:
+            available = list(self.compound_info["MolWt"].keys())
</code_context>

<issue_to_address>
**suggestion:** Raising ValueError instead of BioyondException may lead to inconsistent error handling.

Use BioyondException here for consistency, unless ValueError is required for a specific reason.

```suggestion
            raise BioyondException(
                f"以下化合物未在 compound_info 中定义: {undefined_compounds}。"
                f"可用的化合物: {available}"
            )
```
</issue_to_address>

### Comment 2
<location> `unilabos/devices/workstation/bioyond_studio/dispensing_station.py:304-309` </location>
<code_context>
+        diamine_compounds = [(name, ratio_val, mw, i) for i, (name, ratio_val, mw, fg) in enumerate(ordered_compounds) if fg == "Amine"]
+        anhydride_compounds = [(name, ratio_val, mw, i) for i, (name, ratio_val, mw, fg) in enumerate(ordered_compounds) if fg == "Anhydride"]
+
+        if not diamine_compounds or not anhydride_compounds:
+            raise ValueError(
+                f"需要同时包含二胺(Amine)和二酐(Anhydride)化合物。"
</code_context>

<issue_to_address>
**suggestion:** Error message for missing compound types could be more actionable.

Including the input ratio in the error message will make it easier for users to identify which input caused the issue.

```suggestion
        if not diamine_compounds or not anhydride_compounds:
            diamine_info = [(c[0], c[1]) for c in diamine_compounds]
            anhydride_info = [(c[0], c[1]) for c in anhydride_compounds]
            raise ValueError(
                f"需要同时包含二胺(Amine)和二酐(Anhydride)化合物。\n"
                f"当前二胺: {diamine_info} (格式: 名称, 摩尔比), "
                f"当前二酐: {anhydride_info} (格式: 名称, 摩尔比)\n"
                f"输入的原始摩尔比分别为: {compound_ratios}"
            )
```
</issue_to_address>

### Comment 3
<location> `unilabos/devices/workstation/bioyond_studio/dispensing_station.py:444-447` </location>
<code_context>
+        })
+
+        # 4. 补加溶剂
+        if m_solvent_add > 0:
+            feeding_order.append({
+                "step": len(feeding_order) + 1,
</code_context>

<issue_to_address>
**suggestion (bug_risk):** Potential floating point comparison issue with m_solvent_add > 0.

Use an epsilon value when comparing floating point numbers to zero to prevent precision errors.

```suggestion
        # 4. 补加溶剂
        epsilon = 1e-8
        if m_solvent_add > epsilon:
            feeding_order.append({
                "step": len(feeding_order) + 1,
```
</issue_to_address>

### Comment 4
<location> `unilabos/devices/workstation/bioyond_studio/dispensing_station.py:484` </location>
<code_context>
+            "minimum_required_mass": m_tot_min.magnitude
+        }
+
+        return results
+
     # 90%10%小瓶投料任务创建方法
</code_context>

<issue_to_address>
**suggestion (bug_risk):** Unit magnitudes are extracted for output, but input units are not validated.

Validate input units to ensure consistency and avoid errors from unexpected unit types.

Suggested implementation:

```python
        return results

    # 90%10%小瓶投料任务创建方法
    def create_90_10_vial_feeding_task(self,
                                       order_name: str = None,
                                       m_solvent_titration=None,
                                       m_solvent_add=None,
                                       total_liquid_volume=None,
                                       m_tot_min=None,
                                       *args, **kwargs):
        # Validate input units
        from pint import Quantity

        expected_units = {
            "m_solvent_titration": "gram",
            "m_solvent_add": "gram",
            "total_liquid_volume": "milliliter",
            "m_tot_min": "gram"
        }

        for var_name, expected_unit in expected_units.items():
            var = locals()[var_name]
            if not isinstance(var, Quantity):
                raise ValueError(f"{var_name} must be a Pint Quantity with units of {expected_unit}.")
            if not var.check(expected_unit):
                raise ValueError(f"{var_name} must have units of {expected_unit}, got {var.units}.")


```

You may need to adjust the function signature to ensure all relevant inputs are explicitly passed and validated. If these variables are set elsewhere, move the validation to where they are first assigned.
If you use a different units library or have a custom Quantity class, replace the Pint-specific checks accordingly.
</issue_to_address>

### Comment 5
<location> `unilabos/devices/workstation/bioyond_studio/dispensing_station.py:250` </location>
<code_context>
    def _generate_experiment_design(
        self,
        ratio: dict,
        wt_percent: float = 0.25,
        m_tot: float = 70,
        titration_percent: float = 0.03,
    ) -> dict:
        """内部方法:生成实验设计

        根据FuncGroup自动区分二胺和二酐,每种二胺单独配溶液,严格按照ratio顺序投料。

        参数:
            ratio: 化合物配比字典,格式: {"compound_name": ratio_value}
            wt_percent: 固体重量百分比
            m_tot: 反应混合物总质量(g)
            titration_percent: 滴定溶液百分比

        返回:
            包含实验设计详细参数的字典
        """
        # 溶剂密度
        ρ_solvent = 1.03 * self.ureg.g / self.ureg.ml
        # 二酐溶解度
        solubility = 0.02 * self.ureg.g / self.ureg.ml
        # 投入固体时最小溶剂体积
        V_min = 30 * self.ureg.ml
        m_tot = m_tot * self.ureg.g

        # 保持ratio中的顺序
        compound_names = list(ratio.keys())
        compound_ratios = list(ratio.values())

        # 验证所有化合物是否在 compound_info 中定义
        undefined_compounds = [name for name in compound_names if name not in self.compound_info["MolWt"]]
        if undefined_compounds:
            available = list(self.compound_info["MolWt"].keys())
            raise ValueError(
                f"以下化合物未在 compound_info 中定义: {undefined_compounds}。"
                f"可用的化合物: {available}"
            )

        # 获取各化合物的分子量和官能团类型
        molecular_weights = [self.compound_info["MolWt"][name] for name in compound_names]
        func_groups = [self.compound_info["FuncGroup"][name] for name in compound_names]

        # 记录化合物信息用于调试
        self.hardware_interface._logger.info(f"化合物名称: {compound_names}")
        self.hardware_interface._logger.info(f"官能团类型: {func_groups}")

        # 按原始顺序分离二胺和二酐
        ordered_compounds = list(zip(compound_names, compound_ratios, molecular_weights, func_groups))
        diamine_compounds = [(name, ratio_val, mw, i) for i, (name, ratio_val, mw, fg) in enumerate(ordered_compounds) if fg == "Amine"]
        anhydride_compounds = [(name, ratio_val, mw, i) for i, (name, ratio_val, mw, fg) in enumerate(ordered_compounds) if fg == "Anhydride"]

        if not diamine_compounds or not anhydride_compounds:
            raise ValueError(
                f"需要同时包含二胺(Amine)和二酐(Anhydride)化合物。"
                f"当前二胺: {[c[0] for c in diamine_compounds]}, "
                f"当前二酐: {[c[0] for c in anhydride_compounds]}"
            )

        # 计算加权平均分子量 (基于摩尔比)
        total_molar_ratio = sum(compound_ratios)
        weighted_molecular_weight = sum(ratio_val * mw for ratio_val, mw in zip(compound_ratios, molecular_weights))

        # 取最后一个二酐用于滴定
        titration_anhydride = anhydride_compounds[-1]
        solid_anhydrides = anhydride_compounds[:-1] if len(anhydride_compounds) > 1 else []

        # 二胺溶液配制参数 - 每种二胺单独配制
        diamine_solutions = []
        total_diamine_volume = 0 * self.ureg.ml

        # 计算反应物的总摩尔量
        n_reactant = m_tot * wt_percent / weighted_molecular_weight

        for name, ratio_val, mw, order_index in diamine_compounds:
            # 跳过 SIDA
            if name == "SIDA":
                continue

            # 计算该二胺需要的摩尔数
            n_diamine_needed = n_reactant * ratio_val

            # 二胺溶液配制参数 (每种二胺固定配制参数)
            m_diamine_solid = 5.0 * self.ureg.g  # 每种二胺固体质量
            V_solvent_for_this = 20 * self.ureg.ml  # 每种二胺溶剂体积
            m_solvent_for_this = ρ_solvent * V_solvent_for_this

            # 计算该二胺溶液的浓度
            c_diamine = (m_diamine_solid / mw) / V_solvent_for_this

            # 计算需要移取的溶液体积
            V_diamine_needed = n_diamine_needed / c_diamine

            diamine_solutions.append({
                "name": name,
                "order": order_index,
                "solid_mass": m_diamine_solid.magnitude,
                "solvent_volume": V_solvent_for_this.magnitude,
                "concentration": c_diamine.magnitude,
                "volume_needed": V_diamine_needed.magnitude,
                "molar_ratio": ratio_val
            })

            total_diamine_volume += V_diamine_needed

        # 按原始顺序排序
        diamine_solutions.sort(key=lambda x: x["order"])

        # 计算滴定二酐的质量
        titration_name, titration_ratio, titration_mw, _ = titration_anhydride
        m_titration_anhydride = n_reactant * titration_ratio * titration_mw
        m_titration_90 = m_titration_anhydride * (1 - titration_percent)
        m_titration_10 = m_titration_anhydride * titration_percent

        # 计算其他固体二酐的质量 (按顺序)
        solid_anhydride_masses = []
        for name, ratio_val, mw, order_index in solid_anhydrides:
            mass = n_reactant * ratio_val * mw
            solid_anhydride_masses.append({
                "name": name,
                "order": order_index,
                "mass": mass.magnitude,
                "molar_ratio": ratio_val
            })

        # 按原始顺序排序
        solid_anhydride_masses.sort(key=lambda x: x["order"])

        # 计算溶剂用量
        total_diamine_solution_mass = sum(
            sol["volume_needed"] * ρ_solvent for sol in diamine_solutions
        ) * self.ureg.ml

        # 预估滴定溶剂量、计算补加溶剂量
        m_solvent_titration = m_titration_10 / solubility * ρ_solvent
        m_solvent_add = m_tot * (1 - wt_percent) - total_diamine_solution_mass - m_solvent_titration

        # 检查最小溶剂体积要求
        total_liquid_volume = (total_diamine_solution_mass + m_solvent_add) / ρ_solvent
        m_tot_min = V_min / total_liquid_volume * m_tot

        # 如果需要,按比例放大
        scale_factor = 1.0
        if m_tot_min > m_tot:
            scale_factor = (m_tot_min / m_tot).magnitude
            m_titration_90 *= scale_factor
            m_titration_10 *= scale_factor
            m_solvent_add *= scale_factor
            m_solvent_titration *= scale_factor

            # 更新二胺溶液用量
            for sol in diamine_solutions:
                sol["volume_needed"] *= scale_factor

            # 更新固体二酐用量
            for anhydride in solid_anhydride_masses:
                anhydride["mass"] *= scale_factor

            m_tot = m_tot_min

        # 生成投料顺序
        feeding_order = []

        # 1. 固体二酐 (按顺序)
        for anhydride in solid_anhydride_masses:
            feeding_order.append({
                "step": len(feeding_order) + 1,
                "type": "solid_anhydride",
                "name": anhydride["name"],
                "amount": anhydride["mass"],
                "order": anhydride["order"]
            })

        # 2. 二胺溶液 (按顺序)
        for sol in diamine_solutions:
            feeding_order.append({
                "step": len(feeding_order) + 1,
                "type": "diamine_solution",
                "name": sol["name"],
                "amount": sol["volume_needed"],
                "order": sol["order"]
            })

        # 3. 主要二酐粉末
        feeding_order.append({
            "step": len(feeding_order) + 1,
            "type": "main_anhydride",
            "name": titration_name,
            "amount": m_titration_90.magnitude,
            "order": titration_anhydride[3]
        })

        # 4. 补加溶剂
        if m_solvent_add > 0:
            feeding_order.append({
                "step": len(feeding_order) + 1,
                "type": "additional_solvent",
                "name": "溶剂",
                "amount": m_solvent_add.magnitude,
                "order": 999
            })

        # 5. 滴定二酐溶液
        feeding_order.append({
            "step": len(feeding_order) + 1,
            "type": "titration_anhydride",
            "name": f"{titration_name} 滴定液",
            "amount": m_titration_10.magnitude,
            "titration_solvent": m_solvent_titration.magnitude,
            "order": titration_anhydride[3]
        })

        # 返回实验设计结果
        results = {
            "total_mass": m_tot.magnitude,
            "scale_factor": scale_factor,
            "solutions": diamine_solutions,
            "solids": solid_anhydride_masses,
            "titration": {
                "name": titration_name,
                "main_portion": m_titration_90.magnitude,
                "titration_portion": m_titration_10.magnitude,
                "titration_solvent": m_solvent_titration.magnitude,
            },
            "solvents": {
                "additional_solvent": m_solvent_add.magnitude,
                "total_liquid_volume": total_liquid_volume.magnitude
            },
            "feeding_order": feeding_order,
            "minimum_required_mass": m_tot_min.magnitude
        }

        return results

</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 assignment with augmented assignment ([`aug-assign`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/aug-assign/))
- Inline variable that is immediately returned ([`inline-immediately-returned-variable`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/inline-immediately-returned-variable/))
- Low code quality found in BioyondDispensingStation.\_generate\_experiment\_design - 22% ([`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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

"minimum_required_mass": m_tot_min.magnitude
}

return results
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (bug_risk): Unit magnitudes are extracted for output, but input units are not validated.

Validate input units to ensure consistency and avoid errors from unexpected unit types.

Suggested implementation:

        return results

    # 90%10%小瓶投料任务创建方法
    def create_90_10_vial_feeding_task(self,
                                       order_name: str = None,
                                       m_solvent_titration=None,
                                       m_solvent_add=None,
                                       total_liquid_volume=None,
                                       m_tot_min=None,
                                       *args, **kwargs):
        # Validate input units
        from pint import Quantity

        expected_units = {
            "m_solvent_titration": "gram",
            "m_solvent_add": "gram",
            "total_liquid_volume": "milliliter",
            "m_tot_min": "gram"
        }

        for var_name, expected_unit in expected_units.items():
            var = locals()[var_name]
            if not isinstance(var, Quantity):
                raise ValueError(f"{var_name} must be a Pint Quantity with units of {expected_unit}.")
            if not var.check(expected_unit):
                raise ValueError(f"{var_name} must have units of {expected_unit}, got {var.units}.")

You may need to adjust the function signature to ensure all relevant inputs are explicitly passed and validated. If these variables are set elsewhere, move the validation to where they are first assigned.
If you use a different units library or have a custom Quantity class, replace the Pint-specific checks accordingly.

改进节点ID解析逻辑以支持多种格式,包括字符串和数字标识符
添加数据类型转换处理,确保写入值时类型匹配
优化错误提示信息,便于调试节点连接问题
添加后处理站的YAML配置文件,包含动作映射、状态类型和设备描述
… configurations

- Removed redundant action value mappings from bioyond_dispensing_station.
- Updated goal properties in bioyond_dispensing_station to use enums for target_stack and other parameters.
- Changed data types for end_point and start_point in reaction_station_bioyond to use string enums (Start, End).
- Simplified descriptions and updated measurement units from μL to mL where applicable.
- Removed unused commands from reaction_station_bioyond to streamline the configuration.
# Conflicts:
#	unilabos/device_comms/opcua_client/client.py
#	unilabos/device_comms/opcua_client/node/uniopcua.py
#	unilabos/registry/devices/post_process_station.yaml
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant