demo lakeshore with simulation
- example for lakeshore tutorial - lakeshore simulator (merge with frappy_psils370sim) - rename stringio-server to sim-server Change-Id: I33a9c75ea268349573f8a8387910921e19f242eb Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/30516 Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de> Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
parent
c8f30582a5
commit
05cf4a791a
@ -34,15 +34,17 @@ Use cases, mainly for test purposes:
|
||||
"""
|
||||
|
||||
import sys
|
||||
import argparse
|
||||
from os import path
|
||||
import asyncore
|
||||
import socket
|
||||
import ast
|
||||
import time
|
||||
|
||||
# Add import path for inplace usage
|
||||
sys.path.insert(0, path.abspath(path.join(path.dirname(__file__), '..')))
|
||||
|
||||
from frappy.lib import get_class, formatException
|
||||
from frappy.lib import get_class, formatException, mkthread
|
||||
|
||||
|
||||
class LineHandler(asyncore.dispatcher_with_send):
|
||||
|
||||
@ -51,6 +53,9 @@ class LineHandler(asyncore.dispatcher_with_send):
|
||||
asyncore.dispatcher_with_send.__init__(self, sock)
|
||||
self.crlf = 0
|
||||
|
||||
def handle_line(self, line):
|
||||
raise NotImplementedError
|
||||
|
||||
def handle_read(self):
|
||||
data = self.recv(8192)
|
||||
if data:
|
||||
@ -74,20 +79,22 @@ class LineHandler(asyncore.dispatcher_with_send):
|
||||
|
||||
class LineServer(asyncore.dispatcher):
|
||||
|
||||
def __init__(self, host, port, lineHandlerClass):
|
||||
def __init__(self, port, line_handler_cls, handler_args):
|
||||
asyncore.dispatcher.__init__(self)
|
||||
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.set_reuse_addr()
|
||||
self.bind((host, port))
|
||||
self.bind(('0.0.0.0', port))
|
||||
self.listen(5)
|
||||
self.lineHandlerClass = lineHandlerClass
|
||||
print('accept connections at port', port)
|
||||
self.line_handler_cls = line_handler_cls
|
||||
self.handler_args = handler_args
|
||||
|
||||
def handle_accept(self):
|
||||
pair = self.accept()
|
||||
if pair is not None:
|
||||
sock, addr = pair
|
||||
print("Incoming connection from %s" % repr(addr))
|
||||
self.lineHandlerClass(sock)
|
||||
self.line_handler_cls(sock, self.handler_args)
|
||||
|
||||
def loop(self):
|
||||
asyncore.loop()
|
||||
@ -108,47 +115,73 @@ class Server(LineServer):
|
||||
|
||||
|
||||
class Handler(LineHandler):
|
||||
def __init__(self, sock, handler_args):
|
||||
super().__init__(sock)
|
||||
self.module = handler_args['module']
|
||||
self.verbose = handler_args['verbose']
|
||||
|
||||
def handle_line(self, line):
|
||||
try:
|
||||
reply = module.do_communicate(line.strip())
|
||||
if verbose:
|
||||
reply = self.module.communicate(line.strip())
|
||||
if self.verbose:
|
||||
print('%-40s | %s' % (line, reply))
|
||||
except Exception:
|
||||
print(formatException(verbose=True))
|
||||
return
|
||||
self.send_line(reply)
|
||||
|
||||
|
||||
class Logger:
|
||||
def debug(self, *args):
|
||||
print(*args)
|
||||
info = exception = debug
|
||||
|
||||
|
||||
opts = {'description': 'simulator'}
|
||||
args = []
|
||||
for arg in sys.argv[1:]:
|
||||
k, sep, v = arg.partition('=')
|
||||
if not k:
|
||||
args.append(v)
|
||||
try:
|
||||
v = ast.literal_eval(v)
|
||||
except Exception:
|
||||
pass
|
||||
opts[k] = v
|
||||
verbose = opts.pop('verbose', False)
|
||||
opts['cls'] = 'frappy_psi.ls370sim.Ls370Sim'
|
||||
opts['port'] = 4567
|
||||
if len(args) > 2:
|
||||
raise ValueError('do not know about: %s' % ' '.join(args[2:]))
|
||||
if len(args) == 2:
|
||||
opts['port'] = int(args[1])
|
||||
if len(args) > 0:
|
||||
opts['cls'] = args[0]
|
||||
args.append(opts)
|
||||
|
||||
cls = opts.pop('cls')
|
||||
port = opts.pop('port')
|
||||
srv = Server('localhost', int(port), Handler)
|
||||
module = get_class(cls)(cls, Logger(), opts, srv)
|
||||
module.earlyInit()
|
||||
srv.loop()
|
||||
def log(self, level, *args):
|
||||
pass
|
||||
|
||||
def info(self, *args):
|
||||
print(*args)
|
||||
|
||||
exception = error = warn = info
|
||||
|
||||
|
||||
def parse_argv(argv):
|
||||
parser = argparse.ArgumentParser(description="Simulate HW with a serial interface")
|
||||
parser.add_argument("-v", "--verbose",
|
||||
help="output full communication",
|
||||
action='store_true', default=False)
|
||||
parser.add_argument("cls",
|
||||
type=str,
|
||||
help="simulator class.\n",)
|
||||
parser.add_argument('-p',
|
||||
'--port',
|
||||
action='store',
|
||||
help='server port or uri',
|
||||
default=2089)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def poller(pollfunc):
|
||||
while True:
|
||||
time.sleep(1.0)
|
||||
pollfunc()
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
if argv is None:
|
||||
argv = sys.argv
|
||||
|
||||
args = parse_argv(argv[1:])
|
||||
|
||||
opts = {'description': 'simulator'}
|
||||
|
||||
handler_args = {'verbose': args.verbose}
|
||||
srv = Server(int(args.port), Handler, handler_args)
|
||||
module = get_class(args.cls)(args.cls, Logger(), opts, srv)
|
||||
handler_args['module'] = module
|
||||
module.earlyInit()
|
||||
mkthread(poller, module.doPoll)
|
||||
srv.loop()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main(sys.argv))
|
||||
@ -18,7 +18,11 @@
|
||||
# Module authors:
|
||||
# Markus Zolliker <markus.zolliker@psi.ch>
|
||||
# *****************************************************************************
|
||||
"""a very simple simulator for a LakeShore Model 370"""
|
||||
"""a very simple simulator for a LakeShore
|
||||
|
||||
Model 370: parameters are stored, but no cryo simulation
|
||||
Model 336: heat exchanger on channel A with loop 1, sample sensor on channel B
|
||||
"""
|
||||
|
||||
from frappy.modules import Communicator
|
||||
|
||||
@ -36,40 +40,88 @@ class Ls370Sim(Communicator):
|
||||
('SCAN?', '3,1'),
|
||||
('*OPC?', '1'),
|
||||
]
|
||||
pollinterval = 1
|
||||
|
||||
CHANNELS = list(range(1, 17))
|
||||
data = ()
|
||||
|
||||
def earlyInit(self):
|
||||
super().earlyInit()
|
||||
self._data = dict(self.OTHER_COMMANDS)
|
||||
self.data = dict(self.OTHER_COMMANDS)
|
||||
for fmt, v in self.CHANNEL_COMMANDS:
|
||||
for chan in range(1,17):
|
||||
self._data[fmt % chan] = v
|
||||
for chan in self.CHANNELS:
|
||||
self.data[fmt % chan] = v
|
||||
|
||||
def communicate(self, command):
|
||||
self.comLog('> %s' % command)
|
||||
# simulation part, time independent
|
||||
for channel in range(1,17):
|
||||
def doPoll(self):
|
||||
super().doPoll()
|
||||
self.simulate()
|
||||
|
||||
def simulate(self):
|
||||
# not really a simulation. just for testing RDGST
|
||||
for channel in self.CHANNELS:
|
||||
_, _, _, _, excoff = self._data['RDGRNG?%d' % channel].split(',')
|
||||
if excoff == '1':
|
||||
self._data['RDGST?%d' % channel] = '6'
|
||||
else:
|
||||
self._data['RDGST?%d' % channel] = '0'
|
||||
|
||||
def communicate(self, command):
|
||||
self.comLog('> %s' % command)
|
||||
|
||||
chunks = command.split(';')
|
||||
reply = []
|
||||
for chunk in chunks:
|
||||
if '?' in chunk:
|
||||
reply.append(self._data[chunk])
|
||||
chunk = chunk.replace('? ', '?')
|
||||
reply.append(self.data[chunk])
|
||||
else:
|
||||
for nqarg in (1,0):
|
||||
for nqarg in (1, 0):
|
||||
if nqarg == 0:
|
||||
qcmd, arg = chunk.split(' ', 1)
|
||||
qcmd += '?'
|
||||
else:
|
||||
qcmd, arg = chunk.split(',', nqarg)
|
||||
qcmd = qcmd.replace(' ', '?', 1)
|
||||
if qcmd in self._data:
|
||||
self._data[qcmd] = arg
|
||||
if qcmd in self.data:
|
||||
self.data[qcmd] = arg
|
||||
break
|
||||
reply = ';'.join(reply)
|
||||
self.comLog('< %s' % reply)
|
||||
return reply
|
||||
|
||||
|
||||
class Ls336Sim(Ls370Sim):
|
||||
CHANNEL_COMMANDS = [
|
||||
('KRDG?%s', '295.0'),
|
||||
('RDGST?%s', '0'),
|
||||
]
|
||||
OTHER_COMMANDS = [
|
||||
('*IDN?', 'LSCI,MODEL370,370184,05302003'),
|
||||
('RANGE?1', '0'),
|
||||
('SETP?1', '0'),
|
||||
('CLIMIT?1', ''),
|
||||
('CSET?1', ''),
|
||||
('CMODE?1', ''),
|
||||
('*OPC?', '1'),
|
||||
]
|
||||
|
||||
CHANNELS = 'ABCD'
|
||||
|
||||
vti = 295
|
||||
sample = 295
|
||||
|
||||
def simulate(self):
|
||||
# simple temperature control on channel A:
|
||||
range_ = int(self.data['RANGE?1'])
|
||||
setp = float(self.data['SETP?1'])
|
||||
if range_:
|
||||
# heater on: approach setpoint with 20 sec time constant
|
||||
self.vti = max(self.vti - 0.1, self.vti + (setp - self.vti) * 0.05)
|
||||
else:
|
||||
# heater off 0.1/sec cool down
|
||||
self.vti = max(1.5, self.vti - 0.1)
|
||||
# sample approaching setpoint with 10 sec time constant, but with some
|
||||
# systematic heat loss towards 150 K
|
||||
self.sample = self.sample + (self.vti + (150 - self.vti) * 0.01 - self.sample) * 0.1
|
||||
self.data['KRDG?A'] = str(round(self.vti, 3))
|
||||
self.data['KRDG?B'] = str(round(self.sample, 3))
|
||||
Loading…
x
Reference in New Issue
Block a user