linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Driver for ThermoFisher A 10

Browse Source 
Change-Id: Ic19ae444c3b4242f3bb1fe83852d4521326d0b9d
This commit is contained in:
Oksana Shliakhtun
2023-04-13 15:14:44 +02:00
parent 7a3cfe9836
commit 5fb1e649ab
5 changed files with 259 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
frappy_psi/lakeshore.py
View File

@ -18,11 +18,11 @@
# Module authors:
# Oksana Shliakhtun <oksana.shliakhtun@psi.ch>
# *****************************************************************************
import math
from frappy.core import Readable, Parameter, IntRange, EnumType, FloatRange, \
StringIO, HasIO, StringType, Property, Writable, Drivable, IDLE, ERROR, \
Attached, StructOf, WARN, Done, BoolType
Attached, StructOf, WARN, Done, BoolType, Enum
from frappy_psi.convergence import HasConvergence
from frappy_psi.mixins import HasOutputModule, HasControlledBy
@ -37,20 +37,30 @@ class Ls340IO(StringIO):
class LakeShore(HasIO):
def set_par(self, cmd, *args):
head = ','.join([cmd] + [f'{a:g}' for a in args])
head = ','.join([cmd] + [a if isinstance(a, str) else f'{a:g}' for a in args])
tail = cmd.replace(' ', '? ')
reply = self.communicate(f'{head};{tail}')
reply = [float(num) for num in reply.split(',')]
if len(reply) == 1:
return reply[0]
return reply
result = []
for num in reply.split(','):
try:
result.append(float(num))
except ValueError:
result.append(num)
if len(result) == 1:
return result[0]
return result
def get_par(self, cmd):
reply = self.communicate(cmd)
reply = [float(num) for num in reply.split(',')]
if len(reply) == 1:
return reply[0]
return reply
result = []
for num in reply.split(','):
try:
result.append(float(num))
except ValueError:
result.append(num)
if len(result) == 1:
return result[0]
return result
class Sensor340(LakeShore, Readable):
@ -60,7 +70,7 @@ class Sensor340(LakeShore, Readable):
ioClass = Ls340IO
channel = Property('lakeshore channel', StringType())
alarm = Parameter('alarm limit', FloatRange(unit='K'), readonly=False)
# # define or alter the parameters
# define or alter the parameters
# as Readable.value exists already, we give only the modified property 'unit'
value = Parameter(unit='K')
@ -96,17 +106,16 @@ class HeaterOutput(LakeShore, HasControlledBy, HasIO, Writable):
max_power = Parameter('max heater power', datatype=FloatRange(0, 100), unit='W', readonly=False)
value = Parameter('heater output', datatype=FloatRange(0, 100), unit='W')
target = Parameter('manual heater output', datatype=FloatRange(0, 100), unit='W')
loop = Property('lakeshore loop', IntRange(1, 2), default=1)
channel = Property('attached channel', StringType())
resistance = Property('heater resistance', datatype=FloatRange(10, 1000))
current = Property('heater current', datatype=FloatRange(0, 2))
loop = Property('lakeshore loop', IntRange(1, 2), default=1) # output
channel = Property('attached channel', StringType()) # input
resistance = Property('heater resistance', datatype=FloatRange(10, 100))
_range = 0
_max_power = 50
MAXCURRENTS = {1: 0.25, 2: 0.5, 3: 1.0, 4: 2.0}
RANGES = {1: 1e4, 2: 1e3, 3: 1e2, 4: 1e1, 5: 1}
SETPOINTLIMS = 999
max_current = 0
STATUS_MAP = {
0: (IDLE, ''),
@ -136,8 +145,7 @@ class HeaterOutput(LakeShore, HasControlledBy, HasIO, Writable):
self._range = irange
self.set_par(f'CLIMIT {self.loop}', self.SETPOINTLIMS, 0, 0, icurrent, irange)
self.set_par(f'RANGE {irange}')
self.set_par(f'CDISP {self.loop}', 1, self.resistance, 0)
return self.read_max_power()
self.set_par(f'CDISP {self.loop}', 1, self.resistance, 1, 0)
def read_max_power(self):
setplimit, _, _, icurrent, irange = self.get_par(f'CLIMIT? {self.loop}')
@ -163,7 +171,7 @@ class HeaterOutput(LakeShore, HasControlledBy, HasIO, Writable):
def write_target(self, target):
self.self_controlled()
self.write_max_power(self.max_power)
self.set_heater_mode(3)
self.set_heater_mode(3) # 3=open loop
self.set_range()
percent = self.power_to_percent(target)
reply = self.set_par(f'MOUT {self.loop}', percent)
@ -175,30 +183,91 @@ class HeaterOutput(LakeShore, HasControlledBy, HasIO, Writable):
return self.get_par(f'RANGE?')
def read_value(self):
return self.percent_to_power(self.get_par(f'HTR?'))
return self.percent_to_power(self.get_par(f'HTR?{self.loop}'))
class HeaterOutput340(HeaterOutput):
resistance = Property('heater resistance', datatype=FloatRange(10, 100))
MAXCURRENTS = {1: 0.25, 2: 0.5, 3: 1.0, 4: 2.0}
RANGES = {1: 1e4, 2: 1e3, 3: 1e2, 4: 1e1, 5: 1}
STATUS_MAP = {
0: (IDLE, ''),
1: (ERROR, 'Power supply over voltage'),
2: (ERROR, 'Power supply under voltage'),
3: (ERROR, 'Output digital-to-analog Converter error'),
4: (ERROR, 'Current limit digital-to-analog converter error'),
5: (ERROR, 'Open heater load'),
6: (ERROR, 'Heater load less than 10 ohms')
}
def read_value(self):
return self.percent_to_power(self.get_par(f'HTR?')) # no loop to be given on 340
class HeaterOutput336(HeaterOutput):
power = 20
STATUS_MAP = {
0: (IDLE, ''),
1: (ERROR, 'Open heater load'),
2: (ERROR, 'Heater short')
}
def write_max_power(self, max_power):
max_current = min(math.sqrt(self.power / self.resistance), 2500 / self.resistance)
if self.loop == 1:
max_current_limit = 2
else:
max_current_limit = 1.414
if max_current > max_current_limit:
raise RangeError('max_power above limit')
if max_current >= max_current_limit / math.sqrt(10):
self._range = 3
user_current = max_current
elif max_current >= max_current_limit / 10:
self._range = 2
user_current = max_current * math.sqrt(10)
else:
self._range = 1
user_current = max_current * math.sqrt(100)
self.set_par(f'HTRSET {self.loop}', <1 or 2>, 0, user_current, 1)
max_power = max_current ** 2 * self.resistance
self._max_power = max_power
self.set_range()
return max_power
class TemperatureLoop340(HasConvergence, HasOutputModule, Sensor340, Drivable, LakeShore):
Status = Enum(
Drivable.Status,
RAMPING=370,
STABILIZING=380,
)
target = Parameter(unit='K')
ctrlpars = Parameter('PID parameters',
StructOf(p=FloatRange(0, 1000), i=FloatRange(0, 1000), d=FloatRange(0, 1000)),
readonly=False)
loop = Property('lakeshore loop', IntRange(1, 2), default=1)
ramp = Parameter('ramp rate', FloatRange(min=0, max=1000), unit='K/min', readonly=False)
ramp = Parameter('ramp rate', FloatRange(min=0, max=100), unit='K/min', readonly=False)
ramp_used = Parameter('whether ramp is used or not', BoolType(), readonly=False)
setpoint = Parameter('setpoint', datatype=FloatRange, unit='K')
def doPoll(self):
super().doPoll()
self.read_setpoint()
def write_target(self, target):
out = self.output_module
if out.controlled_by != self.name:
if not self.control_active:
if self.ramp_used:
self.set_par(f'RAMP {self.loop}', 0, self.ramp)
self.set_par(f'SETP {self.loop}', self.value)
self.set_par(f'RAMP {self.loop}', 1, self.ramp)
out.write_target(0)
out.set_heater_mode(1)
out.write_max_power(out.max_power)
out.set_heater_mode(1) # closed loop
self.activate_output()
self.start_state() # start the convergence check
out.set_range()
self.set_par(f'SETP {self.loop}', target)
return target
@ -218,22 +287,37 @@ class TemperatureLoop340(HasConvergence, HasOutputModule, Sensor340, Drivable, L
p, i, d = self.get_par(f'PID? {self.loop}')
return {'p': p, 'i': i, 'd': d}
def write_ramp(self, ramp):
return self.set_par(f'RAMP {self.loop}', self.ramp_used, ramp)[1]
def write_ramp_used(self, ramp_used):
return self.set_par(f'RAMP {self.loop}', ramp_used, self.ramp)[0]
def read_ramp(self):
self.ramp_used, rate = self.get_par(f'RAMP? {self.loop}')
return rate
def write_ramp(self, ramp):
self.ramp_used = True
ramp = self.set_par(f'RAMP {self.loop}', self.ramp_used, ramp)[1]
# if self.control:
# self.ramp = ramp
# self.write_target(self.target)
# return Done
return ramp
def write_ramp_used(self, ramp_used):
ramp_used = self.set_par(f'RAMP {self.loop}', ramp_used, self.ramp)[0]
if self.ramp_used and not ramp_used:
self.write_target(self.target)
return ramp_used
def read_status(self):
statuscode, statustext = super().read_status()
if statuscode == IDLE:
statuscode, statustext = self.status
if self.ramp_used:
if self.read_setpoint() == self.target:
statuscode = self.Status.STABILIZING
else:
statuscode = self.Status.RAMPING
statustext = 'ramping'
if statuscode != ERROR:
return Done
if self.convergence_state.is_active():
self.convergence_state.stop_machine((statuscode, statustext))
if self.convergence_state.is_active:
self.stop_machine((statuscode, statustext))
return ERROR, statustext