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

merged changes for lakeshore and ccu4

Browse Source
This commit is contained in:
l_samenv 2025-03-06 17:26:51 +01:00
parent 95dc8b186e
commit aa64a48ec5
7 changed files with 1565 additions and 371 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
cfg/ls340_cfg.py
View File

@ -4,33 +4,22 @@ Node('ls340test.psi.ch',
)
Mod('io',
'frappy_psi.lakeshore.Ls340IO',
'frappy_psi.lakeshore.IO340',
'communication to ls340',
uri='tcp://ldmprep56-ts:3002'
uri='tcp://localhost:7777'
)
Mod('dev',
'frappy_psi.lakeshore.Device340',
'device for calcurve',
io='io',
curve_handling=True,
)
Mod('T',
'frappy_psi.lakeshore.TemperatureLoop340',
'sample temperature',
output_module='Heater',
target=Param(max=470),
io='io',
channel='B'
)
Mod('T_cold_finger',
'frappy_psi.lakeshore.Sensor340',
'cold finger temperature',
io='io',
channel='A'
)
Mod('Heater',
'frappy_psi.lakeshore.HeaterOutput',
'heater output',
channel='B',
io='io',
resistance=25,
max_power=50,
current=1
'sample temperature',
# output_module='Heater',
device='dev',
channel='A',
calcurve='x29746',
)

17
cfg/main/ori7test_cfg.py Normal file
View File

@ -0,0 +1,17 @@
from frappy_psi.ccracks import Rack
Node('ori7test.psi.ch',
'ORI7 test',
'tcp://5000'
)
rack = Rack(Mod)
with rack.lakeshore() as ls:
ls.sensor('Ts', channel='C', calcurve='x186350')
ls.loop('T', channel='B', calcurve='x174786')
ls.heater('htr', '100W', 100)
rack.ccu(he=True, n2=True)
rack.hepump()

50
frappy/lib/units.py Normal file
View File

@ -0,0 +1,50 @@
# *****************************************************************************
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
# Module authors:
# Markus Zolliker <markus.zolliker@psi.ch>
#
# *****************************************************************************
"""handling of prefixes of physical units"""
import re
import prefixed
prefixed.SI_MAGNITUDE['u'] = 1e-6 # accept 'u' as replacement for 'µ'
class NumberWithUnit:
def __init__(self, *units):
pfx = "|".join(prefixed.SI_MAGNITUDE)
unt = "|".join(units)
self.units = units
self.pattern = re.compile(rf'\s*([+-]?\d*\.?\d*(?:[eE][+-]?\d+)?\s*(?:{pfx})?)({unt})\s*$')
def parse(self, value):
"""parse and return number and value"""
match = self.pattern.match(value)
if not match:
raise ValueError(f'{value!r} can not be interpreted as a number with unit {",".join(self.units)}')
number, unit = match.groups()
return prefixed.Float(number), unit
def getnum(self, value):
"""parse and return value only"""
return self.parse(value)[0]
def format_with_unit(value, unit='', digits=3):
return f'{prefixed.Float(value):.{digits}H}{unit}'

229
frappy_psi/calcurve.py
View File

@ -22,6 +22,7 @@
import os
import re
from pathlib import Path
from os.path import basename, dirname, exists, join
import numpy as np
@ -31,13 +32,22 @@ from scipy.interpolate import PchipInterpolator, CubicSpline, PPoly # pylint: d
from frappy.errors import ProgrammingError, RangeError
from frappy.lib import clamp
def identity(x):
return x
def exp10(x):
return 10 ** np.array(x)
to_scale = {
'lin': lambda x: x,
'log': lambda x: np.log10(x),
'lin': identity,
'log': np.log10,
}
from_scale = {
'lin': lambda x: x,
'log': lambda x: 10 ** np.array(x),
'lin': identity,
'log': exp10,
}
TYPES = [ # lakeshore type, inp-type, loglog
('DT', 'si', False), # Si diode
@ -55,7 +65,7 @@ TYPES = [ # lakeshore type, inp-type, loglog
OPTION_TYPE = {
'loglog': 0, # boolean
'extrange': 2, # tuple(min T, max T for extrapolation
'extrange': 2, # tuple(min T, max T) for extrapolation
'calibrange': 2, # tuple(min T, max T)
}
@ -222,14 +232,6 @@ PARSERS = {
}
def check(x, y, islog):
# check interpolation error
yi = y[:-2] + (x[1:-1] - x[:-2]) * (y[2:] - y[:-2]) / (x[2:] - x[:-2])
if islog:
return sum((yi - y[1:-1]) ** 2)
return sum((np.log10(yi) - np.log10(y[1:-1])) ** 2)
def get_curve(newscale, curves):
"""get curve from curve cache (converts not existing ones)
@ -247,6 +249,7 @@ def get_curve(newscale, curves):
class CalCurve(HasOptions):
EXTRAPOLATION_AMOUNT = 0.1
MAX_EXTRAPOLATION_FACTOR = 2
filename = None # calibration file
def __init__(self, calibspec=None, *, x=None, y=None, cubic_spline=True, **options):
"""calibration curve
@ -257,7 +260,7 @@ class CalCurve(HasOptions):
[<full path> | <name>][,<key>=<value> ...]
for <key>/<value> as in parser arguments
:param x, y: x and y arrays (given instead of calibspec)
:param cubic_split: set to False for always using Pchip interpolation
:param cubic_spline: set to False for always using Pchip interpolation
:param options: options for parsers
"""
self.options = options
@ -265,26 +268,31 @@ class CalCurve(HasOptions):
parser = StdParser()
parser.xdata = x
parser.ydata = y
self.calibname = 'custom'
else:
if x or y:
raise ProgrammingError('can not give both calibspec and x,y ')
sensopt = calibspec.split(',')
calibname = sensopt.pop(0)
_, dot, ext = basename(calibname).rpartition('.')
self.calibname = basename(calibname)
head, dot, ext = self.calibname.rpartition('.')
if dot:
self.calibname = head
kind = None
pathlist = os.environ.get('FRAPPY_CALIB_PATH', '').split(':')
pathlist.append(join(dirname(__file__), 'calcurves'))
pathlist = [Path(p.strip()) for p in os.environ.get('FRAPPY_CALIB_PATH', '').split(':')]
pathlist.append(Path(dirname(__file__)) / 'calcurves')
for path in pathlist:
# first try without adding kind
filename = join(path.strip(), calibname)
if exists(filename):
filename = path / calibname
if filename.exists():
kind = ext if dot else None
break
# then try adding all kinds as extension
for nam in calibname, calibname.upper(), calibname.lower():
for kind in PARSERS:
filename = join(path.strip(), '%s.%s' % (nam, kind))
filename = path / f'{nam}.{kind}'
if exists(filename):
self.filename = filename
break
else:
continue
@ -328,6 +336,7 @@ class CalCurve(HasOptions):
not_incr_idx = np.argwhere(x[1:] <= x[:-1])
if len(not_incr_idx):
raise RangeError('x not monotonic at x=%.4g' % x[not_incr_idx[0]])
self.ptc = y[-1] > y[0]
self.x = {parser.xscale: x}
self.y = {parser.yscale: y}
@ -344,8 +353,7 @@ class CalCurve(HasOptions):
self.convert_x = to_scale[newscale]
self.convert_y = from_scale[newscale]
self.calibrange = self.options.get('calibrange')
dirty = set()
self.extra_points = False
self.extra_points = (0, 0)
self.cutted = False
if self.calibrange:
self.calibrange = sorted(self.calibrange)
@ -371,7 +379,6 @@ class CalCurve(HasOptions):
self.y = {newscale: y}
ibeg = 0
iend = len(x)
dirty.add('xy')
else:
self.extra_points = ibeg, len(x) - iend
else:
@ -493,13 +500,48 @@ class CalCurve(HasOptions):
except IndexError:
return defaultx
def export(self, logformat=False, nmax=199, yrange=None, extrapolate=True, xlimits=None):
def interpolation_error(self, x0, x1, y0, y1, funx, funy, relerror, return_tuple=False):
"""calcualte interpoaltion error
:param x0: start of interval
:param x1: end of interval
:param y0: y at start of interval
:param y1: y at end of interval
:param funx: function to convert x from exported scale to internal scale
:param funy: function to convert y from internal scale to exported scale
:param relerror: True when the exported y scale is linear
:param return_tuple: True: return interpolation error as a tuple with two values
(without and with 3 additional points)
False: return one value without additional points
:return: relative deviation
"""
xspace = np.linspace(x0, x1, 9)
x = funx(xspace)
yr = self.spline(x)
yspline = funy(yr)
yinterp = y0 + np.linspace(0.0, y1 - y0, 9)
# difference between spline (at m points) and liner interpolation
diff = np.abs(yspline - yinterp)
# estimate of interpolation error with 4 sections:
# difference between spline (at m points) and linear interpolation between neighboring points
if relerror:
fact = 2 / (np.abs(y0) + np.abs(y1)) # division by zero can not happen, as y0 and y1 can not both be zero
else:
fact = 2.3 # difference is in log10 -> multiply by 1 / log10(e)
result = np.max(diff, axis=0) * fact
if return_tuple:
diff2 = np.abs(0.5 * (yspline[:-2:2] + yspline[2::2]) - funy(yr[1:-1:2]))
return result, np.max(diff2, axis=0) * fact
return result
def export(self, logformat=False, nmax=199, yrange=None, extrapolate=True, xlimits=None, nmin=199):
"""export curve for downloading to hardware
:param nmax: max number of points. if the number of given points is bigger,
the points with the lowest interpolation error are omitted
:param logformat: a list with two elements of None, True or False
True: use log, False: use line, None: use log if self.loglog
:param logformat: a list with two elements of None, True or False for x and y
True: use log, False: use lin, None: use log if self.loglog
values None are replaced with the effectively used format
False / True are replaced by [False, False] / [True, True]
default is False
@ -507,25 +549,26 @@ class CalCurve(HasOptions):
:param extrapolate: a flag indicating whether the curves should be extrapolated
to the preset extrapolation range
:param xlimits: max x range
:param nmin: minimum number of points
:return: numpy array with 2 dimensions returning the curve
"""
if logformat in (True, False):
logformat = [logformat, logformat]
logformat = (logformat, logformat)
self.logformat = list(logformat)
try:
scales = []
for idx, logfmt in enumerate(logformat):
if logfmt and self.lin_forced[idx]:
raise ValueError('%s must contain positive values only' % 'xy'[idx])
logformat[idx] = linlog = self.loglog if logfmt is None else logfmt
self.logformat[idx] = linlog = self.loglog if logfmt is None else logfmt
scales.append('log' if linlog else 'lin')
xscale, yscale = scales
except (TypeError, AssertionError):
raise ValueError('logformat must be a 2 element list or a boolean')
raise ValueError('logformat must be a 2 element sequence or a boolean')
x = self.spline.x[1:-1] # raw units, excluding extrapolated points
x1, x2 = xmin, xmax = x[0], x[-1]
y1, y2 = sorted(self.spline([x1, x2]))
xr = self.spline.x[1:-1] # raw units, excluding extrapolated points
x1, x2 = xmin, xmax = xr[0], xr[-1]
if extrapolate and not yrange:
yrange = self.exty
@ -535,42 +578,100 @@ class CalCurve(HasOptions):
lim = to_scale[self.scale](xlimits)
xmin = clamp(xmin, *lim)
xmax = clamp(xmax, *lim)
# start and end index of calibrated range
ibeg, iend = self.extra_points[0], len(xr) - self.extra_points[1]
if xmin != x1 or xmax != x2:
ibeg, iend = np.searchsorted(x, (xmin, xmax))
if abs(x[ibeg] - xmin) < 0.1 * (x[ibeg + 1] - x[ibeg]):
i, j = np.searchsorted(xr, (xmin, xmax))
if abs(xr[i] - xmin) < 0.1 * (xr[i + 1] - xr[i]):
# remove first point, if close
ibeg += 1
if abs(x[iend - 1] - xmax) < 0.1 * (x[iend - 1] - x[iend - 2]):
i += 1
if abs(xr[j - 1] - xmax) < 0.1 * (xr[j - 1] - xr[j - 2]):
# remove last point, if close
iend -= 1
x = np.concatenate(([xmin], x[ibeg:iend], [xmax]))
y = self.spline(x)
j -= 1
offset = i - 1
xr = np.concatenate(([xmin], xr[i:j], [xmax]))
ibeg = max(0, ibeg - offset)
iend = min(len(xr), iend - offset)
yr = self.spline(xr)
# convert to exported scale
if xscale != self.scale:
x = to_scale[xscale](from_scale[self.scale](x))
if yscale != self.scale:
y = to_scale[yscale](from_scale[self.scale](y))
# reduce number of points, if needed
n = len(x)
i, j = 1, n - 1 # index range for calculating interpolation deviation
deviation = np.zeros(n)
while True:
# calculate interpolation error when a single point is omitted
ym = y[i-1:j-1] + (x[i:j] - x[i-1:j-1]) * (y[i+1:j+1] - y[i-1:j-1]) / (x[i+1:j+1] - x[i-1:j-1])
if yscale == 'log':
deviation[i:j] = np.abs(ym - y[i:j])
if xscale == self.scale:
xbwd = identity
x = xr
else:
if self.scale == 'log':
xfwd, xbwd = from_scale[self.scale], to_scale[self.scale]
else:
deviation[i:j] = np.abs(ym - y[i:j]) / (np.abs(ym + y[i:j]) + 1e-10)
if n <= nmax:
break
idx = np.argmin(deviation[1:-1]) + 1 # find index of the smallest error
y = np.delete(y, idx)
x = np.delete(x, idx)
deviation = np.delete(deviation, idx)
n -= 1
# index range to recalculate
i, j = max(1, idx - 1), min(n - 1, idx + 1)
self.deviation = deviation # for debugging purposes
xfwd, xbwd = to_scale[xscale], from_scale[xscale]
x = xfwd(xr)
if yscale == self.scale:
yfwd = identity
y = yr
else:
if self.scale == 'log':
yfwd = from_scale[self.scale]
else:
yfwd = to_scale[yscale]
y = yfwd(yr)
self.deviation = None
nmin = min(nmin, nmax)
n = len(x)
relerror = yscale == 'lin'
if len(x) > nmax:
# reduce number of points, if needed
i, j = 1, n - 1 # index range for calculating interpolation deviation
deviation = np.zeros(n)
while True:
deviation[i:j] = self.interpolation_error(
x[i-1:j-1], x[i+1:j+1], y[i-1:j-1], y[i+1:j+1],
xbwd, yfwd, relerror)
# calculate interpolation error when a single point is omitted
if n <= nmax:
break
idx = np.argmin(deviation[1:-1]) + 1 # find index of the smallest error
y = np.delete(y, idx)
x = np.delete(x, idx)
deviation = np.delete(deviation, idx)
n = len(x)
# index range to recalculate
i, j = max(1, idx - 1), min(n - 1, idx + 1)
self.deviation = deviation # for debugging purposes
elif n < nmin:
if ibeg + 1 < iend:
diff1, diff4 = self.interpolation_error(
x[ibeg:iend - 1], x[ibeg + 1:iend], y[ibeg:iend - 1], y[ibeg + 1:iend],
xbwd, yfwd, relerror, return_tuple=True)
dif_target = 1e-4
sq4 = np.sqrt(diff4) * 4
sq1 = np.sqrt(diff1)
offset = 0.49
n_mid = nmax - len(x) + iend - ibeg - 1
# iteration to find a dif target resulting in no more than nmax points
while True:
scale = 1 / np.sqrt(dif_target)
# estimate number of intermediate points (float!) needed to reach dif_target
# number of points estimated from the result of the interpolation error with 4 sections
n4 = np.maximum(1, sq4 * scale)
# number of points estimated from the result of the interpolation error with 1 section
n1 = np.maximum(1, sq1 * scale)
# use n4 where n4 > 4, n1, where n1 < 1 and a weighted average in between
nn = np.select([n4 > 4, n1 > 1],
[n4, (n4 * (n1 - 1) + n1 * (4 - n4)) / (3 + n1 - n4)], n1)
n_tot = np.sum(np.rint(nn + offset))
extra = n_tot - n_mid
if extra <= 0:
break
dif_target *= (n_tot / n_mid) ** 2
xnew = [x[:ibeg]]
for x0, x1, ni in zip(x[ibeg:iend-1], x[ibeg+1:iend], np.rint(nn + offset)):
xnew.append(np.linspace(x0, x1, int(ni) + 1)[:-1])
xnew.append(x[iend-1:])
x = np.concatenate(xnew)
y = yfwd(self.spline(xbwd(x)))
# for debugging purposes:
self.deviation = self.interpolation_error(x[:-1], x[1:], y[:-1], y[1:], xbwd, yfwd, relerror)
return np.stack([x, y], axis=1)

125
frappy_psi/ccracks.py Normal file
View File

@ -0,0 +1,125 @@
import os
from configparser import ConfigParser
class Lsc:
def __init__(self, modfactory, ls_uri, ls_ioname='lsio', ls_devname='ls', ls_model='336', **kwds):
self.modfactory = Mod = modfactory
self.model = ls_model
self.ioname = ls_ioname
self.devname = ls_devname
self.io = Mod(self.ioname, cls=f'frappy_psi.lakeshore.IO{self.model}',
description='comm. to lakeshore in cc rack',
uri=ls_uri)
self.dev = Mod(self.devname, cls=f'frappy_psi.lakeshore.Device{self.model}',
description='lakeshore in cc rack', io=self.ioname, curve_handling=True)
self.loops = {}
self.outputs = {}
def sensor(self, name, channel, calcurve, **kwds):
Mod = self.modfactory
kwds.setdefault('cls', f'frappy_psi.lakeshore.Sensor{self.model}')
kwds.setdefault('description', f'T sensor {name}')
return Mod(name, channel=channel, calcurve=calcurve,
io=self.ioname, device=self.devname, **kwds)
def loop(self, name, channel, calcurve, **kwds):
Mod = self.modfactory
kwds.setdefault('cls', f'frappy_psi.lakeshore.Loop{self.model}')
kwds.setdefault('description', f'T loop {name}')
mod = Mod(name, channel=channel, calcurve=calcurve,
io=self.ioname, device=self.devname, **kwds)
self.loops[name] = mod
return mod
def heater(self, name, max_heater, resistance, output_no=1, **kwds):
Mod = self.modfactory
if output_no == 1:
kwds.setdefault('cls', f'frappy_psi.lakeshore.MainOutput{self.model}')
elif output_no == 2:
kwds.setdefault('cls', f'frappy_psi.lakeshore.SecondaryOutput{self.model}')
else:
return
kwds.setdefault('description', '')
mod = Mod(name, max_heater=max_heater, resistance=resistance,
io=self.ioname, device=self.devname, **kwds)
self.outputs[name] = mod
return mod
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
outmodules = dict(self.outputs)
for name, loop in self.loops.items():
outname = loop.get('output_module')
if outname:
out = outmodules.pop(outname, None)
if not out:
raise KeyError(f'{outname} is not a output module in this lakeshore')
else:
if not outmodules:
raise KeyError(f'{name} needs an output module on this lakeshore')
outname = list(outmodules)[0]
out = outmodules.pop(outname)
loop['output_module'] = outname
if not out['description']:
out['description'] = f'heater for {outname}'
class CCU:
def __init__(self, Mod, ccu_uri, ccu_ioname='ccuio', ccu_devname='ccu', he=None, n2=None, **kwds):
self.ioname = ccu_ioname
self.devname = ccu_devname
Mod(self.ioname, 'frappy_psi.ccu4.CCU4IO',
'comm. to CCU4', uri=ccu_uri)
if he:
if not isinstance(he, str): # e.g. True
he = 'He_lev'
Mod(he, cls='frappy_psi.ccu4.HeLevel',
description='the He Level', io=self.ioname)
if n2:
if isinstance(n2, str):
n2 = n2.split(',')
else: # e.g. True
n2 = []
n2, valve, upper, lower = n2 + ['N2_lev', 'N2_valve', 'N2_upper', 'N2_lower'][len(n2):]
print(n2, valve, upper, lower)
Mod(n2, cls='frappy_psi.ccu4.N2Level',
description='the N2 Level', io=self.ioname,
valve=valve, upper=upper, lower=lower)
Mod(valve, cls='frappy_psi.ccu4.N2FillValve',
description='LN2 fill valve', io=self.ioname)
Mod(upper, cls='frappy_psi.ccu4.N2TempSensor',
description='upper LN2 sensor')
Mod(lower, cls='frappy_psi.ccu4.N2TempSensor',
description='lower LN2 sensor')
class HePump:
def __init__(self, Mod, hepump_uri, hepump_io='hepump_io', hemotname='hepump_mot', **kwds):
Mod(hepump_io, 'frappy_psi.trinamic.BytesIO', 'He pump connection', uri=hepump_uri)
Mod(hemotname, 'frappy_psi.trinamic.Motor', 'He pump valve motor', io=hepump_io)
class Rack:
rackfile = '/home/l_samenv/.config/racks.ini'
def __init__(self, modfactory, **kwds):
self.modfactory = modfactory
parser = ConfigParser()
parser.optionxform = str
parser.read([self.rackfile])
kwds.update(parser.items(os.environ['Instrument']))
self.kwds = kwds
def lakeshore(self):
return Lsc(self.modfactory, **self.kwds)
def ccu(self, **kwds):
kwds.update(self.kwds)
return CCU(self.modfactory, **kwds)
def hepump(self):
return HePump(self.modfactory, **self.kwds)

26
frappy_psi/ccu4.py
View File

@ -28,7 +28,6 @@ from frappy.core import HasIO, Parameter, Command, Readable, Writable, Drivable,
Property, StringIO, BUSY, IDLE, WARN, ERROR, DISABLED, Attached
from frappy.datatypes import BoolType, EnumType, FloatRange, StructOf, \
StatusType, IntRange, StringType, TupleOf
from frappy.dynamic import Pinata
from frappy.errors import CommunicationFailedError
from frappy.states import HasStates, status_code, Retry
@ -42,7 +41,7 @@ class CCU4IO(StringIO):
# for completeness: (not needed, as it is the default)
end_of_line = '\n'
# on connect, we send 'cid' and expect a reply starting with 'CCU4'
identification = [('cid', r'CCU4.*')]
identification = [('cid', r'cid=CCU4.*')]
class CCU4Base(HasIO):
@ -144,7 +143,7 @@ class Valve(CCU4Base, Writable):
self.command(**self._close_command)
def read_status(self):
state = self.command(self._query_state)
state = int(self.command(**self._query_state))
self.value, status = self.STATE_MAP[state]
return status
@ -174,14 +173,14 @@ class N2TempSensor(Readable):
value = Parameter('LN2 T sensor', FloatRange(unit='K'), default=0)
class N2Level(CCU4Base, Pinata, Readable):
class N2Level(CCU4Base, Readable):
valve = Attached(Writable, mandatory=False)
lower = Attached(Readable, mandatory=False)
upper = Attached(Readable, mandatory=False)
value = Parameter('vessel state', EnumType(empty=0, ok=1, full=2))
status = Parameter(datatype=StatusType(Readable, 'BUSY'))
mode = Parameter('auto mode', EnumType(A), readonly=False)
status = Parameter(datatype=StatusType(Readable, 'DISABLED', 'BUSY'))
mode = Parameter('auto mode', EnumType(A), readonly=False, default=A.manual)
threshold = Parameter('threshold triggering start/stop filling',
FloatRange(unit='K'), readonly=False)
@ -206,15 +205,6 @@ class N2Level(CCU4Base, Pinata, Readable):
5: (WARN, 'empty'),
}
def scanModules(self):
for modname, name in self.names.items():
if name:
sensor_name = name.replace('$', self.name)
self.setProperty(modname, sensor_name)
yield sensor_name, {
'cls': N2FillValve if modname == 'valve' else N2TempSensor,
'description': f'LN2 {modname} T sensor'}
def initialReads(self):
self.command(nav=1) # tell CCU4 to activate LN2 sensor readings
super().initialReads()
@ -280,17 +270,19 @@ class N2Level(CCU4Base, Pinata, Readable):
@Command()
def fill(self):
"""start filling"""
self.mode = A.auto
self.io.write(nc=1)
self.command(nc=1)
@Command()
def stop(self):
"""stop filling"""
if self.mode == A.auto:
# set to watching
self.command(nc=3)
else:
# set to off
self.io.write(nc=0)
self.command(nc=0)
class FlowPressure(CCU4Base, Readable):

1452
frappy_psi/lakeshore.py
View File

File diff suppressed because it is too large Load Diff