python3-damaris/src/damaris/tools/lakeshore.py

import serial
import re
import time
import numpy as np

from .temperature import TemperatureController, register_controller


DEBUG = False

reply_pattern = re.compile(r"\x02..(.*)\x03.", re.DOTALL)
# example answer '\x02PV279.8\x03/'
# [EOT] = \x04
# [STX] = \x02
# [ENQ] = \x05
# [ETX] = \x03
# [ACK] = \x06
# BCC = checksum

standard_device = '/dev/ttyUSB0'


"""
Parameter read example:

Master: [EOT]0011PV[ENQ]
Instrument:			[STX]PV16.4[ETX]{BCC}

Writing data:

Master: [EOT] {GID}{GID}{UID}{UID}[STX]{CHAN}(c1)(c2)<DATA>[ETX](BCC)

"""



class Lakeshore(TemperatureController):
    def __init__(self, serial_device, baudrate=57600):
        self.device = standard_device
        # timeout: 110 ms to get all answers.
        self.s = serial.Serial(serial_device,
                               baudrate=baudrate,
                               bytesize=7,
                               parity='O',
                               stopbits=1,
                               timeout=0.4)
        self._expect_len = 50

    def __del__(self):
        if hasattr(self, 's') and self.s is not None:
            if self.s.is_open:
                self.s.close()

    #def get_current_temperature(self):  # Temp A
     #   self.s.write("KRDG? a\n")
     #  a = self.s.readline().strip()[1:]
     #   return float(a)

    def get_temperature_A(self):  # Temp A
        self.s.write("KRDG? a\n")
        a = self.s.readline().strip()[1:]
        return float(a)

    def get_temperature_B(self):  # Temp B
        self.s.write("KRDG? b\n")
        b = self.s.readline().strip()[1:]
        return float(b)

    def set_temperature(self, temperature):     # setze Setpoint
        t = temperature
        self.s.write(f"SETP 1,{t}\n")

    def set_ramp(self, ramp):
        return self.s.write(f'RAMP 1, 1, {ramp}\n')
        
    def get_set_temperature(self):      # Setpoint abfrage
        self.s.write("SETP? 1\n")
        s = self.s.readline().strip()[1:]
        return float(s)

    def get_output(self):               # Heater Output
        self.s.write("HTR? 1\n")
        o = self.s.readline().strip()[1:]
        return float(o)

    def get_ramp(self):
        self.s.write('RAMP? 1\n')
        ret = self.s.readline().strip()
        onoff, rate = ret.split(',')
        if onoff:
            return float(rate)
        else:
            return 0.0

    def read_param(self, param, retries=3, retry_delay=0.05):
        """Read a parameter using simple ASCII commands."""
        for attempt in range(retries):
            self.s.flushInput()
            self.s.write(f"{param}?\n")
            answer = self.s.readline().strip()
            if answer and len(answer) > 1:
                # Remove the first character (command echo) and return the value
                return answer[1:]
            if attempt < retries - 1:
                time.sleep(retry_delay)
        return None

    def get_value(self, name):
        temp = self.read_param(name)
        if temp is None:
            return np.nan
        else:
            try:
                return float(temp)
            except ValueError:
                return np.nan

    # TemperatureController interface implementation

    def get_temperature(self) -> float:
        """Read current temperature in Celsius."""
        temp = self.read_param('KRDG')
        if temp is None:
            return 0.0
        try:
            return float(temp)
        except (ValueError, TypeError):
            return 0.0

    def get_setpoint(self) -> float:
        """Read target setpoint temperature in Celsius."""
        temp = self.read_param('SETP')
        if temp is None:
            return 0.0
        try:
            return float(temp)
        except (ValueError, TypeError):
            return 0.0

    def set_setpoint(self, temperature: float) -> bool:
        """Set target temperature in Celsius."""
        self.s.write(f"SETP 1,{temperature}\n")
        return True

    def check_connection(self) -> bool:
        """Check if the Lakeshore device is connected and responsive."""
        try:
            result = self.read_param('KRDG')
            return result is not None
        except Exception:
            return False

    def close(self):
        """Close the serial connection."""
        if hasattr(self, 's') and self.s is not None:
            if self.s.is_open:
                self.s.close()
            self.s = None


# Register Lakeshore with the factory
register_controller('lakeshore', Lakeshore)