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

add 'ts' to the ppms simulation

Browse Source 
+ convert from CRLF to LF

Change-Id: I46fab0c970ccc5e7e704a5dc0ab2cfd51213cd31
Reviewed-on: https://forge.frm2.tum.de/review/c/sine2020/secop/playground/+/28233
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
This commit is contained in:
zolliker
2022-04-22 16:14:26 +02:00
parent d3379d5e95
commit b7d16d2e16
1 changed files with 237 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

469
secop_psi/ppmssim.py
View File

@ -1,232 +1,237 @@
#!/usr/bin/env python #!/usr/bin/env python
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# ***************************************************************************** # *****************************************************************************
# This program is free software; you can redistribute it and/or modify it under # 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 # 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 # Foundation; either version 2 of the License, or (at your option) any later
# version. # version.
# #
# This program is distributed in the hope that it will be useful, but WITHOUT # 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 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more # FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details. # details.
# #
# You should have received a copy of the GNU General Public License along with # 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., # this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
# #
# Module authors: # Module authors:
# Markus Zolliker <markus.zolliker@psi.ch> # Markus Zolliker <markus.zolliker@psi.ch>
# ***************************************************************************** # *****************************************************************************
import json import json
import math import math
import time import time
def num(string): def num(string):
return json.loads(string) return json.loads(string)
class NamedList: class NamedList:
def __init__(self, keys, *args, **kwargs): def __init__(self, keys, *args, **kwargs):
self.__keys__ = keys.split() self.__keys__ = keys.split()
self.setvalues(args) self.setvalues(args)
for key, val in kwargs.items(): for key, val in kwargs.items():
setattr(self, key, val) setattr(self, key, val)
def setvalues(self, values): def setvalues(self, values):
for key, arg in zip(self.__keys__, values): for key, arg in zip(self.__keys__, values):
setattr(self, key, arg) setattr(self, key, arg)
def aslist(self): def aslist(self):
return [getattr(self, key) for key in self.__keys__] return [getattr(self, key) for key in self.__keys__]
def __getitem__(self, index): def __getitem__(self, index):
return getattr(self, self.__keys__[index]) return getattr(self, self.__keys__[index])
def __setitem__(self, index, value): def __setitem__(self, index, value):
return setattr(self, self.__keys__[index], value) return setattr(self, self.__keys__[index], value)
def __repr__(self): def __repr__(self):
return ",".join("%.7g" % val for val in self.aslist()) return ",".join("%.7g" % val for val in self.aslist())
class PpmsSim: class PpmsSim:
CHANNELS = 'st t mf pos r1 i1 r2 i2'.split() CHANNELS = {
0: 'st', 1: 't', 2: 'mf', 3: 'pos', 4: 'r1', 5: 'i1', 6: 'r2', 7: 'i2',
def __init__(self): 23: 'ts',
self.status = NamedList('t mf ch pos', 1, 1, 1, 1) }
self.st = 0x1111
self.t = 15 def __init__(self):
self.temp = NamedList('target ramp amode', 200., 1, 0, fast=self.t, delay=10) self.status = NamedList('t mf ch pos', 1, 1, 1, 1)
self.mf = 100 self.st = 0x1111
self.field = NamedList('target ramp amode pmode', 0, 50, 0, 0) self.t = 15
self.pos = 0 self.temp = NamedList('target ramp amode', 200., 1, 0, fast=self.t, delay=10)
self.move = NamedList('target mode code', 0, 0, 0) self.mf = 100
self.chamber = NamedList('target', 0) self.field = NamedList('target ramp amode pmode', 0, 50, 0, 0)
self.level = NamedList('value code', 100.0, 1) self.pos = 0
self.bridge1 = NamedList('no exc pow dc mode vol', 1, 333, 1000, 0, 2, 1) self.move = NamedList('target mode code', 0, 0, 0)
self.bridge2 = NamedList('no exc pow dc mode vol', 2, 333, 1000, 0, 2, 1) self.chamber = NamedList('target', 0)
self.bridge3 = NamedList('no exc pow dc mode vol', 3, 333, 1000, 0, 2, 1) self.level = NamedList('value code', 100.0, 1)
self.bridge4 = NamedList('no exc pow dc mode vol', 4, 333, 1000, 0, 2, 1) self.bridge1 = NamedList('no exc pow dc mode vol', 1, 333, 1000, 0, 2, 1)
self.drvout1 = NamedList('no cur pow', 1, 333, 1000) self.bridge2 = NamedList('no exc pow dc mode vol', 2, 333, 1000, 0, 2, 1)
self.drvout2 = NamedList('no cur pow', 2, 333, 1000) self.bridge3 = NamedList('no exc pow dc mode vol', 3, 333, 1000, 0, 2, 1)
self.r1 = 0 self.bridge4 = NamedList('no exc pow dc mode vol', 4, 333, 1000, 0, 2, 1)
self.i1 = 0 self.drvout1 = NamedList('no cur pow', 1, 333, 1000)
self.r2 = 0 self.drvout2 = NamedList('no cur pow', 2, 333, 1000)
self.i2 = 0 self.r1 = 0
self.time = int(time.time()) self.i1 = 0
self.start = self.time self.r2 = 0
self.mf_start = 0 self.i2 = 0
self.ch_start = 0 self.ts = self.t + 0.1
self.t_start = 0 self.time = int(time.time())
self.changed = set() self.start = self.time
self.mf_start = 0
def progress(self): self.ch_start = 0
now = time.time() self.t_start = 0
if self.time >= now: self.changed = set()
return
while self.time < now: def progress(self):
self.time += 1 now = time.time()
if self.temp.amode: # no overshoot if self.time >= now:
dif = self.temp.target - self.temp.fast return
else: while self.time < now:
dif = self.temp.target - self.t self.time += 1
self.temp.fast += math.copysign(min(self.temp.ramp / 60.0, abs(dif)), dif) if self.temp.amode: # no overshoot
self.t += (self.temp.fast - self.t) / self.temp.delay dif = self.temp.target - self.temp.fast
else:
# handle magnetic field dif = self.temp.target - self.t
if 'FIELD' in self.changed: self.temp.fast += math.copysign(min(self.temp.ramp / 60.0, abs(dif)), dif)
self.changed.remove('FIELD') self.t += (self.temp.fast - self.t) / self.temp.delay
if self.field.target < 0:
self.status.mf = 15 # general error # handle magnetic field
elif self.status.mf == 1: # persistent if 'FIELD' in self.changed:
self.mf_start = now # indicates leads are ramping self.changed.remove('FIELD')
elif self.status.mf == 3: # switch_cooling if self.field.target < 0:
self.mf_start = now self.status.mf = 15 # general error
self.status.mf = 2 # switch_warming elif self.status.mf == 1: # persistent
else: self.mf_start = now # indicates leads are ramping
self.status.mf = 6 + int(self.field.target < self.mf) # charging or discharging elif self.status.mf == 3: # switch_cooling
if self.status.mf == 1 and self.mf_start: # leads ramping self.mf_start = now
if now > self.mf_start + abs(self.field.target) / 10000 + 5: self.status.mf = 2 # switch_warming
self.mf_start = now else:
self.status.mf = 2 # switch_warming self.status.mf = 6 + int(self.field.target < self.mf) # charging or discharging
elif self.status.mf == 2: # switch_warming if self.status.mf == 1 and self.mf_start: # leads ramping
if now > self.mf_start + 15: if now > self.mf_start + abs(self.field.target) / 10000 + 5:
self.status.mf = 6 + int(self.field.target < self.mf) # charging or discharging self.mf_start = now
elif self.status.mf == 5: # driven_final self.status.mf = 2 # switch_warming
if now > self.mf_start + 5: elif self.status.mf == 2: # switch_warming
self.mf_start = now if now > self.mf_start + 15:
self.status.mf = 3 # switch cooling self.status.mf = 6 + int(self.field.target < self.mf) # charging or discharging
elif self.status.mf == 3: # switch_cooling elif self.status.mf == 5: # driven_final
if now > self.mf_start + 15: if now > self.mf_start + 5:
self.status.mf = 1 # persistent_mode self.mf_start = now
self.mf_start = 0 # == no leads ramping happens self.status.mf = 3 # switch cooling
elif self.status.mf in (6, 7): # charging, discharging elif self.status.mf == 3: # switch_cooling
dif = self.field.target - self.mf if now > self.mf_start + 15:
if abs(dif) < 0.01: self.status.mf = 1 # persistent_mode
if self.field.pmode: self.mf_start = 0 # == no leads ramping happens
self.status.mf = 4 # driven_stable elif self.status.mf in (6, 7): # charging, discharging
else: dif = self.field.target - self.mf
self.status.mf = 5 # driven_final if abs(dif) < 0.01:
self.mf_last = now if self.field.pmode:
else: self.status.mf = 4 # driven_stable
self.mf += math.copysign(min(self.field.ramp, abs(dif)), dif) else:
# print(self.mf, self.status.mf, self.field) self.status.mf = 5 # driven_final
self.mf_last = now
dif = self.move.target - self.pos else:
speed = (15 - self.move.code) * 0.8 self.mf += math.copysign(min(self.field.ramp, abs(dif)), dif)
self.pos += math.copysign(min(speed, abs(dif)), dif) # print(self.mf, self.status.mf, self.field)
if 'CHAMBER' in self.changed: dif = self.move.target - self.pos
self.changed.remove('CHAMBER') speed = (15 - self.move.code) * 0.8
if self.chamber.target == 0: # seal immediately self.pos += math.copysign(min(speed, abs(dif)), dif)
self.status.ch = 3 # sealed unknown
self.ch_start = 0 if 'CHAMBER' in self.changed:
elif self.chamber.target == 3: # pump cont. self.changed.remove('CHAMBER')
self.status.ch = 8 if self.chamber.target == 0: # seal immediately
self.ch_start = 0 self.status.ch = 3 # sealed unknown
elif self.chamber.target == 4: # vent cont. self.ch_start = 0
self.status.ch = 9 elif self.chamber.target == 3: # pump cont.
self.ch_start = 0 self.status.ch = 8
elif self.chamber.target == 1: # purge and seal self.ch_start = 0
self.status.ch = 4 elif self.chamber.target == 4: # vent cont.
self.ch_start = now self.status.ch = 9
elif self.chamber.target == 2: # vent and seal self.ch_start = 0
self.status.ch = 5 elif self.chamber.target == 1: # purge and seal
self.ch_start = now self.status.ch = 4
elif self.chamber.target == 5: # hi vac. self.ch_start = now
self.status.ch = 6 # pumping down elif self.chamber.target == 2: # vent and seal
self.ch_start = now self.status.ch = 5
elif self.ch_start and now > self.ch_start + 15: self.ch_start = now
self.ch_start = 0 elif self.chamber.target == 5: # hi vac.
if self.chamber.target == 5: self.status.ch = 6 # pumping down
self.status.ch = 7 # at high vac. self.ch_start = now
else: elif self.ch_start and now > self.ch_start + 15:
self.status.ch = self.chamber.target self.ch_start = 0
if self.chamber.target == 5:
if 'TEMP' in self.changed: self.status.ch = 7 # at high vac.
self.changed.remove('TEMP') else:
self.status.t = 2 # changing self.status.ch = self.chamber.target
self.t_start = now
elif abs(self.t - self.temp.target) < 0.1: if 'TEMP' in self.changed:
if now > self.t_start + 10: self.changed.remove('TEMP')
self.status.t = 1 # stable self.status.t = 2 # changing
else: self.t_start = now
self.status.t = 5 # within tolerance elif abs(self.t - self.temp.target) < 0.1:
else: if now > self.t_start + 10:
self.t_start = now self.status.t = 1 # stable
if abs(self.t - self.temp.target) < 1: else:
self.status.t = 6 # outside tolerance self.status.t = 5 # within tolerance
else:
if abs(self.pos - self.move.target) < 0.01: self.t_start = now
self.status.pos = 1 if abs(self.t - self.temp.target) < 1:
else: self.status.t = 6 # outside tolerance
self.status.pos = 5
if abs(self.pos - self.move.target) < 0.01:
self.st = sum([self.status[i] << (i * 4) for i in range(4)]) self.status.pos = 1
self.r1 = self.t * 0.1 else:
self.i1 = self.t % 10.0 self.status.pos = 5
self.r2 = 1000 / self.t
self.i2 = math.log(self.t) self.st = sum([self.status[i] << (i * 4) for i in range(4)])
self.level.value = 100 - (self.time - self.start) * 0.01 % 100 self.r1 = self.t * 0.1
self.i1 = self.t % 10.0
def getdat(self, mask): self.r2 = 1000 / self.t
mask = int(mask) & 0xff # all channels up to i2 self.i2 = math.log(self.t)
output = ['%d' % mask, '%.2f' % (time.time() - self.start)] self.ts = self.t + 0.1
for i, chan in enumerate(self.CHANNELS): self.level.value = 100 - (self.time - self.start) * 0.01 % 100
if (1 << i) & mask:
output.append("%.7g" % getattr(self, chan)) def getdat(self, mask):
return ",".join(output) mask = int(mask) & 0x8000ff # all channels up to i2 plus ts
output = ['%d' % mask, '%.2f' % (time.time() - self.start)]
for i, chan in self.CHANNELS.items():
class QDevice: if (1 << i) & mask:
def __init__(self, classid): output.append("%.7g" % getattr(self, chan))
self.sim = PpmsSim() return ",".join(output)
def send(self, command):
self.sim.progress() class QDevice:
if '?' in command: def __init__(self, classid):
if command.startswith('GETDAT?'): self.sim = PpmsSim()
mask = int(command[7:])
result = self.sim.getdat(mask) def send(self, command):
else: self.sim.progress()
name, args = command.split('?') if '?' in command:
name += args.strip() if command.startswith('GETDAT?'):
result = getattr(self.sim, name.lower()).aslist() mask = int(command[7:])
result = ",".join("%.7g" % arg for arg in result) result = self.sim.getdat(mask)
# print(command, '/', result) else:
else: name, args = command.split('?')
# print(command) name += args.strip()
name, args = command.split() result = getattr(self.sim, name.lower()).aslist()
args = json.loads("[%s]" % args) result = ",".join("%.7g" % arg for arg in result)
if name.startswith('BRIDGE') or name.startswith('DRVOUT'): # print(command, '/', result)
name = name + str(int(args[0])) else:
getattr(self.sim, name.lower()).setvalues(args) # print(command)
self.sim.changed.add(name) name, args = command.split()
result = "OK" args = json.loads("[%s]" % args)
return result if name.startswith('BRIDGE') or name.startswith('DRVOUT'):
name = name + str(int(args[0]))
getattr(self.sim, name.lower()).setvalues(args)
def shutdown(): self.sim.changed.add(name)
pass result = "OK"
return result
def shutdown():
pass