z-stage leveling complete

Author: andy

__version__.py

@@ -1,7 +1,7 @@
 import os
 import subprocess
 
-VERSION = '1.1.4'
+VERSION = '1.1.5'
 
 
 def get_version():

dist/Productions.exe

@@ -11,7 +11,7 @@ class CalibrateMethod(enum.Enum):
 
     Mount.LEFT: {
 
-        "Z-C2": {"point": Point(195, 197, 357), "compensation": {"rear": 0, "front": 0},
+        "Z-C2": {"point": Point(215, 197, 357), "compensation": {"rear": 0, "front": 0},
                  "channel_definition": ["below_rear", "below_front"]},
 
     },

ot3_testing/test_config/zstage_leveling_config.py

@@ -6,8 +6,11 @@
 from ot3_testing.test_config.zstage_leveling_config import ZStagePoint, CalibrateMethod, ZStageChannel
 from typing import List
 from drivers.play_sound import play_alarm_3
-
+from utils import Utils
+import os
+import asyncio
 RequestReadyFlag = False
+ApplyCompensationFlag = True
 
 
 class ZStageLeveling(TestBase):
@@ -178,11 +181,11 @@ async def adjust_leveling(self, slot_name: str, mount: Mount):
         input("Judging complete ? （完成校准回车）")
         self.judge_complete = True
 
-    async def run_z_stage_test(self):
+    async def run_z_stage_test(self, project_path=None):
         """
         main loop
         """
-        test_result = {}
+        self.judge_complete = False
         if self.robot_ip is None:
             addr = self.get_address().strip()
         else:
@@ -193,13 +196,16 @@ async def run_z_stage_test(self):
 
         # adjust
         await self.adjust_leveling('Z-C2', Mount.RIGHT)
-
-        await self.api.home()
-
         input("Run Test (开始测试)？")
         # run test
+        await self.api.home()
+
+        csv_title = []
+        csv_list = []
+
         for mount in [Mount.RIGHT, Mount.LEFT]:
             self.mount = mount
+            test_result = {}
             print(f"Start {self.mount.value} side...")
             for p_key, p_value in ZStagePoint[self.mount].items():
                 _point = p_value["point"]
@@ -208,12 +214,47 @@ async def run_z_stage_test(self):
                 result = await self.run_test_slot(_point, p_key, _channel_definition)
                 test_result.update(result)
             await self.api.home()
-        print(test_result)
-        # save
+            print(test_result)
+            # save
+            for key, value in test_result.items():
+                result = []
+                distance_list = list(value.values())
+                difference = round(abs(distance_list[0] - distance_list[1]), 3)
+                compensation = ZStagePoint[self.mount][key]["compensation"]
+                if ApplyCompensationFlag:
+                    compensation_idx = 0
+                    for compensation_key, compensation_value in compensation.items():
+                        print(
+                            f"apply offset {compensation_key} -> {compensation_value}  to {distance_list[compensation_idx]}")
+                        result.append(compensation_value + distance_list[compensation_idx])
+                        compensation_idx += 1
+                    difference = round(abs(result[0] - result[1]), 3)
+                print(f"{key} --> {value} (mm) --> difference: {difference}(mm)")
+                for item_key, item_value in value.items():
+                    csv_title.append(self.mount.name + " " + key + " " + item_key)
+                    csv_list.append(item_value)
+                csv_title.append(key + "-Result")
+                csv_list.append(difference)
 
+        if project_path is not None:
+            file_path = os.path.join(project_path, 'testing_data', 'z_stage_leveling.csv')
+        else:
+            file_path = '../../testing_data/z_stage_leveling.csv'
+        self.save_csv(file_path, csv_title, csv_list)
+        self.laser_sensor.close()
 
-if __name__ == '__main__':
-    import asyncio
+    def save_csv(self, file_path, title, content):
+        """
+        save csv
+        """
+        is_exist = Utils.is_file_exist(file_path)
+        if is_exist:
+            pass
+        else:
+            Utils.write_to_csv(file_path, title)
+        Utils.write_to_csv(file_path, content)
 
-    obj = ZStageLeveling(ZStagePoint, robot_ip="192.168.6.51")
+
+if __name__ == '__main__':
+    obj = ZStageLeveling(ZStagePoint, robot_ip="192.168.6.33")
     asyncio.run(obj.run_z_stage_test())

ot3_testing/tests/zstage_leveling.py

@@ -1,8 +1,10 @@
-import os, sys
-
+import os
+import sys
 from ot3_testing.tests.pipette_leveling import PipetteLeveling
+from ot3_testing.tests.zstage_leveling import ZStageLeveling
 from ot3_testing.test_config.pipette_leveling_config import SlotLocationCH96, ChannelDefinitionCH96, \
     SlotLocationCH8, ChannelDefinitionCH8
+from ot3_testing.test_config.zstage_leveling_config import ZStagePoint
 import asyncio
 from tools.inquirer import prompt_flex_name, prompt_test_name, prompt_exit
 from tools import heat_96ch
@@ -30,6 +32,10 @@
         elif "leveling-96ch" in test_name:
             pipette_leveling = PipetteLeveling(SlotLocationCH96, ChannelDefinitionCH96, )
             asyncio.run(pipette_leveling.run_96ch_test(flex_name, project_path=project_path))
+        elif "leveling-z-stage" in test_name:
+            z_leveling = ZStageLeveling(ZStagePoint)
+            asyncio.run(z_leveling.run_z_stage_test(project_path=project_path))
+
         elif "leveling-8ch" in test_name:
             # run 8
             pipette_leveling = PipetteLeveling(SlotLocationCH8, ChannelDefinitionCH8)

production_scripts.py

@@ -1,6 +1,6 @@
 from PyInquirer import prompt
 
-test_choices = ['leveling-8ch', 'leveling-96ch', 'leveling-gantry', 'heat-96ch','grav-openweb']
+test_choices = ['leveling-8ch', 'leveling-96ch', 'leveling-gantry', 'leveling-z-stage', 'heat-96ch', 'grav-openweb']
 
 question_flex = {
     'type': 'input',
@@ -22,7 +22,6 @@
 
 }
 
-
 question_ip = {
     'type': 'input',
     'name': 'ip',
@@ -60,6 +59,7 @@ def prompt_raspNo():
     ret = prompt(question_gravname)
     return ret['raspNo'].strip()
 
+
 def prompt_ip():
     ret = prompt(question_ip)
     return ret['ip'].strip()
@@ -79,7 +79,7 @@ def prompt_exit():
     ret = prompt(question_exit)
     return ret['exit'].strip()
 
+
 def prompt_openweb():
     ret = prompt(question_openweb)
     return ret['openweb'].strip()
-