Public Access
merged pyporject.toml
This commit is contained in:
+1
-1
@@ -20,7 +20,7 @@ dependencies = [
|
||||
"matplotlib >= 1.4.1, <= 3.6",
|
||||
"pyxdg",
|
||||
"PyGObject >= 3.14.0, <= 3.51.0",
|
||||
"scipy-stubs~=1.17.1",
|
||||
"pyserial>=3.5",
|
||||
]
|
||||
authors = [
|
||||
{name = "Achim Gädke"},
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
import serial
|
||||
import re
|
||||
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):
|
||||
"""Read a parameter using simple ASCII commands."""
|
||||
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:]
|
||||
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 {temperature}\n")
|
||||
return True
|
||||
|
||||
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)
|
||||
@@ -44,6 +44,7 @@ def create_controller(name: str, *args, **kwargs):
|
||||
module_map = {
|
||||
'simulated': 'damaris.tools.temperature_simulator',
|
||||
'eurotherm': 'damaris.tools.eurotherm',
|
||||
'lakeshore': 'damaris.tools.lakeshore',
|
||||
}
|
||||
if name in module_map:
|
||||
try:
|
||||
@@ -79,6 +80,13 @@ except ImportError as e:
|
||||
# That's OK, it will be imported on-demand if needed
|
||||
pass
|
||||
|
||||
try:
|
||||
from . import lakeshore
|
||||
except ImportError as e:
|
||||
# Lakeshore may require serial module which might not be installed
|
||||
# That's OK, it will be imported on-demand if needed
|
||||
pass
|
||||
|
||||
try:
|
||||
from . import temperature_simulator
|
||||
except ImportError as e:
|
||||
|
||||
Reference in New Issue
Block a user