Accretech wafer prober driver

src/Temperature-Accretech_UFseries/libs/accretech_uf.py

@@ -34,10 +34,12 @@
 
 import inspect
 from datetime import datetime
+import time
 
 import pysweepme.Ports
-from pysweepme.ErrorMessage import debug
+from pysweepme.ErrorMessage import debug, error
 from pyvisa import constants
+import pyvisa
 
 try:
     from pysweepme.UserInterface import get_input
@@ -184,7 +186,8 @@ def __init__(self, port: pysweepme.Ports.Port):
 
     def __del__(self) -> None:
         """When the object is deleted, the SRQ event is unregistered."""
-        self.unregister_srq_event()
+        pass
+        # self.unregister_srq_event()
 
     def register_srq_event(self) -> None:
         """Register Service Request Events."""
@@ -215,7 +218,7 @@ def get_prober_status_message(self, prober_status: str) -> str:
 
     @staticmethod
     def get_waferlist_from_status(wafer_status: str) -> list[tuple[int, int]]:
-        """Returns a list of tuples with the cassette and wafer numbers."""
+        """Returns a list of tuples with cassette number, wafer number, and status value."""
         wafer_list = []
 
         # iterates through cassettes
@@ -227,7 +230,7 @@ def get_waferlist_from_status(wafer_status: str) -> list[tuple[int, int]]:
                 for wafer_id, val in enumerate(wafers):
                     if val != "0":
                         # adding a tuple of cassette and wafer number
-                        wafer_list.append((cassette_id + 1, wafer_id + 1))
+                        wafer_list.append((cassette_id + 1, wafer_id + 1, int(val)))
 
         return wafer_list
 
@@ -254,7 +257,8 @@ def wait_until_status_byte(self, stb_success: int | tuple, timeout: float = 5.0)
         stb = None
         while stb not in stb_success:
             stb = self.acquire_status_byte(timeout)
-
+            if stb == 99:  # Abnormal end always indicate that we can break here
+                break
         return stb
 
     def acquire_status_byte(self, timeout: float = 10.0) -> int:
@@ -275,7 +279,19 @@ def acquire_status_byte(self, timeout: float = 10.0) -> int:
                 print("-->", now.strftime("%H:%M:%S"), "Function:", function_calling_name)
 
         # Wait for the SRQ event to occur
-        response = self.port.port.wait_on_event(self.event_type, int(timeout * 1000))  # conversion from s to ms
+        starttime = time.time()
+        while True: 
+            if time.time()-starttime < timeout:
+                try:
+                    # response = self.port.port.wait_on_event(self.event_type, int(timeout * 1000))  # conversion from s to ms
+                    response = self.port.port.wait_on_event(self.event_type, int(1000))  # waiting 1000 ms
+                    break
+                except pyvisa.errors.VisaIOError:
+                    # Timeout error
+                    pass
+            else:
+                msg = "Timeout reached during waiting for status byte"
+                raise Exception(msg)
 
         # The event mechanism was changed after pyvisa 1.9.0 and the WaitResponse structure is different
         if hasattr(response, "event"):
@@ -320,7 +336,8 @@ def acquire_status_byte(self, timeout: float = 10.0) -> int:
 
             while True:
                 msg = "Accretech: " + error_type + " (%s): " % error_code + error_message + " after status byte 76"
-                answer = get_input(msg + "\n\nDo you like to continue y/n?")
+                answer = get_input(msg + "\n\nTo continue please confirm with 'y' and then handle the problem at the "
+                                    "prober\nDo you like to continue y/n?")
 
                 if answer.lower() == "y":
                     # we call this function again to make sure a new status byte is retrieved
@@ -632,15 +649,15 @@ def unload(self) -> int:
             # 71
         """
         self.port.write("U")
-        return self.wait_until_status_byte(71, timeout=180.0)
+        return self.wait_until_status_byte(71, timeout=300.0)
 
     def unload_all_wafers(self) -> int:
         """Returns:
         int: status byte
             # 94
         """
         self.port.write("U0")
-        return self.wait_until_status_byte(94, timeout=30.0)
+        return self.wait_until_status_byte((71, 94), timeout=300.0)
 
     def unload_to_inspection_tray(self) -> int:
         """Returns:
@@ -651,7 +668,9 @@ def unload_to_inspection_tray(self) -> int:
         return self.wait_until_status_byte(71, timeout=120.0)
 
     def load_specified_wafer(self, cassette, slot) -> int:
-        """This command can be used to load a wafer from a cassette. It indirectly starts a lot process and is not used with 'start'.
+        """This command can be used to load a wafer from a cassette.
+
+        It indirectly starts a lot process and is not used with 'start'.
 
         You can terminate the lot process by using cassette = 9 and slot = 99
 
@@ -681,7 +700,7 @@ def load_specified_wafer(self, cassette, slot) -> int:
             raise Exception(msg)
 
         self.port.write("j2%i%02d" % (int(cassette), int(slot)))
-        return self.wait_until_status_byte((70, 94), timeout=600.0)
+        return self.wait_until_status_byte((70, 94), timeout=300.0)
 
     def preload_specified_wafer(self, cassette, slot) -> int:
         """This command ("j3") can be used to preload a wafer from a cassette to the subchuck.
@@ -745,6 +764,13 @@ def load_and_preload_specified_wafers(self, cassette, slot, preload_cassette, pr
         self.port.write("j4%i%02d%i%02d" % (int(cassette), int(slot), int(preload_cassette), int(preload_slot)))
         return self.wait_until_status_byte(70, timeout=300.0)
 
+    def enable_reexecution(self):
+        """
+        Enables the re-execution of a lot process.
+        """
+        self.port.write("ji")
+        return self.wait_until_status_byte((98, 99), timeout=60.0)
+
     def load_wafer_aligned(self) -> int:
         """Returns:
         int: status byte
@@ -1011,9 +1037,34 @@ def set_chuck_temperature(self, temperature: float) -> None:
             msg = f"Accretech UF series: Temperature must be between {min_temperature} and {max_temperature} °C."
             raise Exception(msg)
 
-        self.port.write("h" + str(int(temperature * 10)))
+        self.port.write("h%04d" % (temperature * 10))
         answer = self.wait_until_status_byte((93, 99))
 
         correct_status = 93
         if answer != correct_status:  # Abnormal end of command  # noqa: PLR2004
-            self.raise_error(answer)
+            self.raise_error(answer)
+
+    def request_device_name_list(self, storage="c"):
+        """Requests the list of names of saved device parameter sets
+
+        Args:
+            storage: str
+
+        Returns:
+            str: list of all available device parameter set names
+        """
+
+        answer = self.query("d" + storage)
+        return answer
+
+    def request_cassette_lock_status(self):
+        """Requests cassette lock status
+
+        Returns:
+            int:
+                0: cassette is in the unlock status
+                1: cassette is in the lock status
+        """
+
+        answer = self.query("cls")
+        return int(answer)

src/Temperature-Accretech_UFseries/main.py

@@ -33,6 +33,7 @@
 import os
 
 from pysweepme.EmptyDeviceClass import EmptyDevice
+from pysweepme.ErrorMessage import debug
 from pysweepme.FolderManager import addFolderToPATH
 
 addFolderToPATH()
@@ -87,19 +88,24 @@ def __init__(self) -> None:
             "GPIB_EOLread": "\r\n",
         }
 
+        self.port_str = "undefined_port"
+
         self.verbosemode = True
 
     def set_GUIparameter(self) -> dict:  # noqa: N802
         """Set initial GUI parameters."""
         return {
             "SweepMode": ["None", "Temperature"],
             "TemperatureUnit": ["°C"],
+            "IdleTemperature": 30,
         }
 
     def get_GUIparameter(self, parameter: dict) -> None:  # noqa: N802
         """Handle SweepMe GUI parameters."""
+        self.port_str = parameter["Port"]
         self.units[0] = parameter["TemperatureUnit"]
         self.port_str = parameter["Port"]
+        self.idle_temperature = parameter["IdleTemperature"]
 
     def connect(self) -> None:
         """Connect to the device and handle multiple instances."""
@@ -113,26 +119,46 @@ def connect(self) -> None:
         #     self.device_communication[self.instance_key] = self.port
 
         self.prober = accretech_uf.AccretechProber(self.port)
-
         self.prober.set_verbose(self.verbosemode)
 
+        self.unique_identifier = "Driver_Accretech_UFseries_" + self.port_str
+        if not self.unique_identifier in self.device_communication:
+            self.device_communication[self.unique_identifier] = None
+
     def disconnect(self) -> None:
         """Disconnect from the device."""
-        del self.prober  # this makes sure the event mechanism is disabled
+        if self.unique_identifier in self.device_communication:
+            self.prober.unregister_srq_event()  # this makes sure the event mechanism is disabled
+            del self.device_communication[self.unique_identifier]
+        del self.prober  
 
     def initialize(self) -> None:
         """Initialize the device."""
-        # TODO: Check
         self.prober.reset_alarm()
 
+    def deinitialize(self):
+        pass
+
+    def unconfigure(self):
+        if self.idle_temperature != "":
+            _, target_temperature = self.prober.request_chuck_temperature()
+            if target_temperature != float(self.idle_temperature):
+                self.prober.set_chuck_temperature(float(self.idle_temperature))
+            else:
+                debug("Target temperature already set.")
+
     def apply(self) -> None:
         """Apply the given temperature."""
-        self.prober.set_chuck_temperature(self.value)
+        _, target_temperature = self.prober.request_chuck_temperature()
+        if target_temperature != float(self.value):
+            self.prober.set_chuck_temperature(self.value)
+        else:
+            debug("Target temperature already set.")
 
     def measure(self) -> None:
         """Measure the temperature."""
         # TODO: chuck_temperature or hot_chuck_temperature?
-        self.temperature, target_temperature = self.prober.get_temperature()
+        self.temperature, target_temperature = self.prober.request_chuck_temperature()
 
     def call(self) -> list[float]:
         """Return measured temperature to SweepMe GUI."""
@@ -142,5 +168,5 @@ def call(self) -> list[float]:
 
     def measure_temperature(self) -> float:
         """Measure the temperature."""
-        _temperature, _target_temperature = self.prober.get_temperature()
+        _temperature, _target_temperature = self.prober.request_chuck_temperature()
         return _temperature