even more merges from gerrit
Change-Id: I4cfddc8fd4157ceae353789f2f60d834ec05974e
This commit is contained in:
@ -42,7 +42,6 @@ class Ls370Sim(Communicator):
|
||||
for fmt, v in self.CHANNEL_COMMANDS:
|
||||
for chan in range(1,17):
|
||||
self._data[fmt % chan] = v
|
||||
# mkthread(self.run)
|
||||
|
||||
def communicate(self, command):
|
||||
self.comLog('> %s' % command)
|
||||
|
@ -168,7 +168,7 @@ class Channel(PpmsBase):
|
||||
datatype=IntRange(1, 4), export=False)
|
||||
|
||||
def earlyInit(self):
|
||||
Readable.earlyInit(self)
|
||||
super().earlyInit()
|
||||
if not self.channel:
|
||||
self.channel = self.name
|
||||
|
||||
|
@ -26,6 +26,7 @@ import time
|
||||
def num(string):
|
||||
return json.loads(string)
|
||||
|
||||
|
||||
class NamedList:
|
||||
def __init__(self, keys, *args, **kwargs):
|
||||
self.__keys__ = keys.split()
|
||||
@ -49,8 +50,10 @@ class NamedList:
|
||||
def __repr__(self):
|
||||
return ",".join("%.7g" % val for val in self.aslist())
|
||||
|
||||
|
||||
class PpmsSim:
|
||||
CHANNELS = 'st t mf pos r1 i1 r2 i2'.split()
|
||||
|
||||
def __init__(self):
|
||||
self.status = NamedList('t mf ch pos', 1, 1, 1, 1)
|
||||
self.st = 0x1111
|
||||
@ -176,7 +179,6 @@ class PpmsSim:
|
||||
if abs(self.t - self.temp.target) < 1:
|
||||
self.status.t = 6 # outside tolerance
|
||||
|
||||
|
||||
if abs(self.pos - self.move.target) < 0.01:
|
||||
self.status.pos = 1
|
||||
else:
|
||||
@ -187,8 +189,7 @@ class PpmsSim:
|
||||
self.i1 = self.t % 10.0
|
||||
self.r2 = 1000 / self.t
|
||||
self.i2 = math.log(self.t)
|
||||
self.level.value = round(100 - (self.time - self.start) * 0.01 % 100, 1)
|
||||
# print('PROGRESS T=%.7g B=%.7g x=%.7g' % (self.t, self.mf, self.pos))
|
||||
self.level.value = 100 - (self.time - self.start) * 0.01 % 100
|
||||
|
||||
def getdat(self, mask):
|
||||
mask = int(mask) & 0xff # all channels up to i2
|
||||
@ -198,6 +199,7 @@ class PpmsSim:
|
||||
output.append("%.7g" % getattr(self, chan))
|
||||
return ",".join(output)
|
||||
|
||||
|
||||
class QDevice:
|
||||
def __init__(self, classid):
|
||||
self.sim = PpmsSim()
|
||||
@ -225,5 +227,6 @@ class QDevice:
|
||||
result = "OK"
|
||||
return result
|
||||
|
||||
|
||||
def shutdown():
|
||||
pass
|
||||
|
@ -22,12 +22,12 @@
|
||||
|
||||
import math
|
||||
import os
|
||||
from os.path import basename, exists, join
|
||||
from os.path import basename, dirname, exists, join
|
||||
|
||||
import numpy as np
|
||||
from scipy.interpolate import splev, splrep # pylint: disable=import-error
|
||||
|
||||
from secop.core import Attached, BoolType, Parameter, Readable, StringType
|
||||
from secop.core import Attached, BoolType, Parameter, Readable, StringType, FloatRange
|
||||
|
||||
|
||||
def linear(x):
|
||||
@ -74,13 +74,18 @@ class Parser340(StdParser):
|
||||
def parse(self, line):
|
||||
"""scan header for data format"""
|
||||
if self.header:
|
||||
if line.startswith("Data Format"):
|
||||
dataformat = line.split(":")[1].strip()[0]
|
||||
if dataformat == '4':
|
||||
self.logx, self.logy = True, False # logOhm
|
||||
elif dataformat == '5':
|
||||
self.logx, self.logy = True, True # logOhm, logK
|
||||
elif line.startswith("No."):
|
||||
key, _, value = line.partition(':')
|
||||
if value: # this is a header line, as it contains ':'
|
||||
value = value.split()[0]
|
||||
key = ''.join(key.split()).lower()
|
||||
if key == 'dataformat':
|
||||
if value == '4':
|
||||
self.logx, self.logy = True, False # logOhm
|
||||
elif value == '5':
|
||||
self.logx, self.logy = True, True # logOhm, logK
|
||||
elif value not in ('1', '2', '3'):
|
||||
raise ValueError('invalid Data Format')
|
||||
elif 'No.' in line:
|
||||
self.header = False
|
||||
return
|
||||
super().parse(line)
|
||||
@ -104,7 +109,9 @@ class CalCurve:
|
||||
calibname = sensopt.pop(0)
|
||||
_, dot, ext = basename(calibname).rpartition('.')
|
||||
kind = None
|
||||
for path in os.environ.get('FRAPPY_CALIB_PATH', '').split(','):
|
||||
pathlist = os.environ.get('FRAPPY_CALIB_PATH', '').split(',')
|
||||
pathlist.append(join(dirname(__file__), 'calcurves'))
|
||||
for path in pathlist:
|
||||
# first try without adding kind
|
||||
filename = join(path.strip(), calibname)
|
||||
if exists(filename):
|
||||
@ -134,13 +141,26 @@ class CalCurve:
|
||||
cls, args = KINDS.get(kind, (StdParser, {}))
|
||||
args.update(optargs)
|
||||
|
||||
parser = cls(**args)
|
||||
with open(filename) as f:
|
||||
for line in f:
|
||||
parser.parse(line)
|
||||
try:
|
||||
parser = cls(**args)
|
||||
with open(filename) as f:
|
||||
for line in f:
|
||||
parser.parse(line)
|
||||
except Exception as e:
|
||||
raise ValueError('calib curve %s: %s' % (calibspec, e)) from e
|
||||
self.convert_x = nplog if parser.logx else linear
|
||||
self.convert_y = npexp if parser.logy else linear
|
||||
self.spline = splrep(np.asarray(parser.xdata), np.asarray(parser.ydata), s=0)
|
||||
x = np.asarray(parser.xdata)
|
||||
y = np.asarray(parser.ydata)
|
||||
if np.all(x[:-1] > x[1:]): # all decreasing
|
||||
x = np.flip(x)
|
||||
y = np.flip(y)
|
||||
elif np.any(x[:-1] >= x[1:]): # some not increasing
|
||||
raise ValueError('calib curve %s is not monotonic' % calibspec)
|
||||
try:
|
||||
self.spline = splrep(x, y, s=0, k=min(3, len(x) - 1))
|
||||
except (ValueError, TypeError) as e:
|
||||
raise ValueError('invalid calib curve %s' % calibspec) from e
|
||||
|
||||
def __call__(self, value):
|
||||
"""convert value
|
||||
@ -156,7 +176,7 @@ class Sensor(Readable):
|
||||
|
||||
calib = Parameter('calibration name', datatype=StringType(), readonly=False)
|
||||
abs = Parameter('True: take abs(raw) before calib', datatype=BoolType(), readonly=False, default=True)
|
||||
value = Parameter(unit='K')
|
||||
value = Parameter(datatype=FloatRange(unit='K'))
|
||||
pollinterval = Parameter(export=False)
|
||||
status = Parameter(default=(Readable.Status.ERROR, 'unintialized'))
|
||||
|
||||
@ -164,9 +184,17 @@ class Sensor(Readable):
|
||||
_value_error = None
|
||||
enablePoll = False
|
||||
|
||||
def checkProperties(self):
|
||||
if 'description' not in self.propertyValues:
|
||||
self.description = '_' # avoid complaining about missing description
|
||||
super().checkProperties()
|
||||
|
||||
def initModule(self):
|
||||
super().initModule()
|
||||
self._rawsensor.registerCallbacks(self, ['status']) # auto update status
|
||||
self._calib = CalCurve(self.calib)
|
||||
if self.description == '_':
|
||||
self.description = '%r calibrated with curve %r' % (self.rawsensor, self.calib)
|
||||
|
||||
def write_calib(self, value):
|
||||
self._calib = CalCurve(value)
|
||||
@ -174,7 +202,7 @@ class Sensor(Readable):
|
||||
|
||||
def update_value(self, value):
|
||||
if self.abs:
|
||||
value = abs(value)
|
||||
value = abs(float(value))
|
||||
self.value = self._calib(value)
|
||||
self._value_error = None
|
||||
|
||||
|
Reference in New Issue
Block a user