add plr_to_bioyond, and refactor bioyond stations

unilabos/devices/workstation/bioyond_studio/reaction_station.py

@@ -0,0 +1,207 @@
+import json
+
+from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation
+from unilabos.devices.workstation.bioyond_studio.config import (
+    API_CONFIG, WORKFLOW_MAPPINGS, WORKFLOW_STEP_IDS, MATERIAL_TYPE_MAPPINGS,
+    STATION_TYPES, DEFAULT_STATION_CONFIG
+)
+
+
+class BioyondReactionStation(BioyondWorkstation):
+    def __init__(self, config: dict = None):
+        super().__init__(config)
+
+    # 工作流方法
+    def reactor_taken_out(self):
+        """反应器取出"""
+        self.hardware_interface.append_to_workflow_sequence('{"web_workflow_name": "reactor_taken_out"}')
+        reactor_taken_out_params = {"param_values": {}}
+        self.hardware_interface.pending_task_params.append(reactor_taken_out_params)
+        print(f"成功添加反应器取出工作流")
+        print(f"当前队列长度: {len(self.hardware_interface.pending_task_params)}")
+        return json.dumps({"suc": True})
+
+    def reactor_taken_in(self, assign_material_name: str, cutoff: str = "900000", temperature: float = -10.00):
+        """反应器放入"""
+        self.append_to_workflow_sequence('{"web_workflow_name": "reactor_taken_in"}')
+        material_id = self._get_material_id_by_name(assign_material_name)
+
+        if isinstance(temperature, str):
+            temperature = float(temperature)
+
+        step_id = WORKFLOW_STEP_IDS["reactor_taken_in"]["config"]
+        reactor_taken_in_params = {
+            "param_values": {
+                step_id: [
+                    {"m": 0, "n": 3, "Key": "cutoff", "Value": cutoff},
+                    {"m": 0, "n": 3, "Key": "temperature", "Value": f"{temperature:.2f}"},
+                    {"m": 0, "n": 3, "Key": "assignMaterialName", "Value": material_id}
+                ]
+            }
+        }
+
+        self.pending_task_params.append(reactor_taken_in_params)
+        print(f"成功添加反应器放入参数: material={assign_material_name}->ID:{material_id}, cutoff={cutoff}, temp={temperature:.2f}")
+        print(f"当前队列长度: {len(self.pending_task_params)}")
+        return json.dumps({"suc": True})
+
+    def solid_feeding_vials(self, material_id: str, time: str = "0", torque_variation: str = "1",
+                            assign_material_name: str = None, temperature: float = 25.00):
+        """固体进料小瓶"""
+        self.append_to_workflow_sequence('{"web_workflow_name": "Solid_feeding_vials"}')
+        material_id_m = self._get_material_id_by_name(assign_material_name)
+
+        if isinstance(temperature, str):
+            temperature = float(temperature)
+
+        feeding_id = WORKFLOW_STEP_IDS["solid_feeding_vials"]["feeding"]
+        observe_id = WORKFLOW_STEP_IDS["solid_feeding_vials"]["observe"]
+
+        solid_feeding_vials_params = {
+            "param_values": {
+                feeding_id: [
+                    {"m": 0, "n": 3, "Key": "materialId", "Value": material_id},
+                    {"m": 0, "n": 3, "Key": "assignMaterialName", "Value": material_id_m}
+                ],
+                observe_id: [
+                    {"m": 1, "n": 0, "Key": "time", "Value": time},
+                    {"m": 1, "n": 0, "Key": "torqueVariation", "Value": torque_variation},
+                    {"m": 1, "n": 0, "Key": "temperature", "Value": f"{temperature:.2f}"}
+                ]
+            }
+        }
+
+        self.pending_task_params.append(solid_feeding_vials_params)
+        print(f"成功添加固体进料小瓶参数: material_id={material_id}, time={time}min, temp={temperature:.2f}°C")
+        print(f"当前队列长度: {len(self.pending_task_params)}")
+        return json.dumps({"suc": True})
+
+    def liquid_feeding_vials_non_titration(self, volumeFormula: str, assign_material_name: str,
+                                          titration_type: str = "1", time: str = "0",
+                                          torque_variation: str = "1", temperature: float = 25.00):
+        """液体进料小瓶(非滴定)"""
+        self.append_to_workflow_sequence('{"web_workflow_name": "Liquid_feeding_vials(non-titration)"}')
+        material_id = self._get_material_id_by_name(assign_material_name)
+
+        if isinstance(temperature, str):
+            temperature = float(temperature)
+
+        liquid_id = WORKFLOW_STEP_IDS["liquid_feeding_vials_non_titration"]["liquid"]
+        observe_id = WORKFLOW_STEP_IDS["liquid_feeding_vials_non_titration"]["observe"]
+
+        params = {
+            "param_values": {
+                liquid_id: [
+                    {"m": 0, "n": 3, "Key": "volumeFormula", "Value": volumeFormula},
+                    {"m": 0, "n": 3, "Key": "assignMaterialName", "Value": material_id},
+                    {"m": 0, "n": 3, "Key": "titrationType", "Value": titration_type}
+                ],
+                observe_id: [
+                    {"m": 1, "n": 0, "Key": "time", "Value": time},
+                    {"m": 1, "n": 0, "Key": "torqueVariation", "Value": torque_variation},
+                    {"m": 1, "n": 0, "Key": "temperature", "Value": f"{temperature:.2f}"}
+                ]
+            }
+        }
+
+        self.pending_task_params.append(params)
+        print(f"成功添加液体进料小瓶(非滴定)参数: volume={volumeFormula}μL, material={assign_material_name}->ID:{material_id}")
+        print(f"当前队列长度: {len(self.pending_task_params)}")
+        return json.dumps({"suc": True})
+
+    def liquid_feeding_solvents(self, assign_material_name: str, volume: str, titration_type: str = "1",
+                               time: str = "360", torque_variation: str = "2", temperature: float = 25.00):
+        """液体进料溶剂"""
+        self.append_to_workflow_sequence('{"web_workflow_name": "Liquid_feeding_solvents"}')
+        material_id = self._get_material_id_by_name(assign_material_name)
+
+        if isinstance(temperature, str):
+            temperature = float(temperature)
+
+        liquid_id = WORKFLOW_STEP_IDS["liquid_feeding_solvents"]["liquid"]
+        observe_id = WORKFLOW_STEP_IDS["liquid_feeding_solvents"]["observe"]
+
+        params = {
+            "param_values": {
+                liquid_id: [
+                    {"m": 0, "n": 1, "Key": "titrationType", "Value": titration_type},
+                    {"m": 0, "n": 1, "Key": "volume", "Value": volume},
+                    {"m": 0, "n": 1, "Key": "assignMaterialName", "Value": material_id}
+                ],
+                observe_id: [
+                    {"m": 1, "n": 0, "Key": "time", "Value": time},
+                    {"m": 1, "n": 0, "Key": "torqueVariation", "Value": torque_variation},
+                    {"m": 1, "n": 0, "Key": "temperature", "Value": f"{temperature:.2f}"}
+                ]
+            }
+        }
+
+        self.pending_task_params.append(params)
+        print(f"成功添加液体进料溶剂参数: material={assign_material_name}->ID:{material_id}, volume={volume}μL")
+        print(f"当前队列长度: {len(self.pending_task_params)}")
+        return json.dumps({"suc": True})
+
+    def liquid_feeding_titration(self, volume_formula: str, assign_material_name: str, titration_type: str = "1",
+                                time: str = "90", torque_variation: int = 2, temperature: float = 25.00):
+        """液体进料(滴定)"""
+        self.append_to_workflow_sequence('{"web_workflow_name": "Liquid_feeding(titration)"}')
+        material_id = self._get_material_id_by_name(assign_material_name)
+
+        if isinstance(temperature, str):
+            temperature = float(temperature)
+
+        liquid_id = WORKFLOW_STEP_IDS["liquid_feeding_titration"]["liquid"]
+        observe_id = WORKFLOW_STEP_IDS["liquid_feeding_titration"]["observe"]
+
+        params = {
+            "param_values": {
+                liquid_id: [
+                    {"m": 0, "n": 3, "Key": "volumeFormula", "Value": volume_formula},
+                    {"m": 0, "n": 3, "Key": "titrationType", "Value": titration_type},
+                    {"m": 0, "n": 3, "Key": "assignMaterialName", "Value": material_id}
+                ],
+                observe_id: [
+                    {"m": 1, "n": 0, "Key": "time", "Value": time},
+                    {"m": 1, "n": 0, "Key": "torqueVariation", "Value": str(torque_variation)},
+                    {"m": 1, "n": 0, "Key": "temperature", "Value": f"{temperature:.2f}"}
+                ]
+            }
+        }
+
+        self.pending_task_params.append(params)
+        print(f"成功添加液体进料滴定参数: volume={volume_formula}μL, material={assign_material_name}->ID:{material_id}")
+        print(f"当前队列长度: {len(self.pending_task_params)}")
+        return json.dumps({"suc": True})
+
+    def liquid_feeding_beaker(self, volume: str = "35000", assign_material_name: str = "BAPP",
+                             time: str = "0", torque_variation: str = "1", titrationType: str = "1",
+                             temperature: float = 25.00):
+        """液体进料烧杯"""
+        self.append_to_workflow_sequence('{"web_workflow_name": "liquid_feeding_beaker"}')
+        material_id = self._get_material_id_by_name(assign_material_name)
+
+        if isinstance(temperature, str):
+            temperature = float(temperature)
+
+        liquid_id = WORKFLOW_STEP_IDS["liquid_feeding_beaker"]["liquid"]
+        observe_id = WORKFLOW_STEP_IDS["liquid_feeding_beaker"]["observe"]
+
+        params = {
+            "param_values": {
+                liquid_id: [
+                    {"m": 0, "n": 2, "Key": "volume", "Value": volume},
+                    {"m": 0, "n": 2, "Key": "assignMaterialName", "Value": material_id},
+                    {"m": 0, "n": 2, "Key": "titrationType", "Value": titrationType}
+                ],
+                observe_id: [
+                    {"m": 1, "n": 0, "Key": "time", "Value": time},
+                    {"m": 1, "n": 0, "Key": "torqueVariation", "Value": torque_variation},
+                    {"m": 1, "n": 0, "Key": "temperature", "Value": f"{temperature:.2f}"}
+                ]
+            }
+        }
+
+        self.pending_task_params.append(params)
+        print(f"成功添加液体进料烧杯参数: volume={volume}μL, material={assign_material_name}->ID:{material_id}")
+        print(f"当前队列长度: {len(self.pending_task_params)}")
+        return json.dumps({"suc": True})