further work on superfrappy
This commit is contained in:
@@ -5,7 +5,7 @@ doc="""super frappy
|
||||
server to keep track of secop servers on an instrument
|
||||
"""
|
||||
Node('superfrappy.psi.ch', doc,
|
||||
interface='tcp://10767',
|
||||
interface='tcp://5001',
|
||||
)
|
||||
|
||||
IO('nicosio', 'localhost:14869')
|
||||
@@ -13,14 +13,13 @@ IO('nicosio', 'localhost:14869')
|
||||
Mod('local',
|
||||
'frappy_psi.superfrappy.SuperFrappy',
|
||||
'this instrument',
|
||||
marchecfgdir = '/home/linse/marchecfg',
|
||||
instrument = 'zebra',
|
||||
main_port = 15101,
|
||||
stick_port = 15201,
|
||||
io = 'nicosio',
|
||||
wrapperdir = '/home/linse/config',
|
||||
setupdir = '/home/linse/setuptest',
|
||||
cfgdirs = '/sq_sw/linse/frappy_zebra/cfg',
|
||||
main_instrument = True,
|
||||
wrapperdir = '/home/linse/cfgfrappy',
|
||||
setupdir = '/home/linse/setups',
|
||||
cfgdirs = '/sq_sw/linse/frappycfg:/sq_sw/linse/frappy_zebra/cfg',
|
||||
target = [],
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
+1
-1
@@ -44,7 +44,7 @@ class JTCCR(Writable):
|
||||
datatype=FloatRange(unit='mbar'), default=1.0)
|
||||
p3margin = Property('start compressor when p3 is below pressreg setpoint plus this value',
|
||||
datatype=FloatRange(unit='mbar'), default=0.01)
|
||||
p3reg = Parameter('pressure regulation setpoint', datatype=FloatRange(unit='mbar'))
|
||||
p3reg = Parameter('pressure regulation setpoint', datatype=FloatRange(unit='mbar'), default=4.0)
|
||||
plow = Property('pressure below 5K', datatype=FloatRange(unit='mbar'), default=4.0)
|
||||
|
||||
valves_high_pressure = {
|
||||
|
||||
+229
-195
@@ -16,13 +16,53 @@
|
||||
# Module authors:
|
||||
# Markus Zolliker <markus.zolliker@psi.ch>
|
||||
# *****************************************************************************
|
||||
"""module to trigger plug and play mechanism
|
||||
|
||||
For all detected secop servers, setup files are written to the
|
||||
setup directory.
|
||||
|
||||
For SEC nodes started on the instrument computer, wrapper config files
|
||||
containing the port number are produced in the wrapper directory.
|
||||
|
||||
Mechanism:
|
||||
|
||||
start or detect a SecNode
|
||||
|
||||
(on start only) create a wrapper file and tell marche to start
|
||||
add SecNode to superfrappy._secnodes in 'connecting' mode
|
||||
|
||||
|
||||
action table:
|
||||
|
||||
connection == online, setupfile does not exist:
|
||||
- announce = new, create wrapperfile (if needed) and setup file
|
||||
|
||||
nicos == idle, setup not in loaded_setups, announce == new:
|
||||
- announce = None, plugplay = new
|
||||
|
||||
setup in loaded_setups, plugplay == new:
|
||||
- pluglay = quiet (handled in _handle_setups)
|
||||
|
||||
connection != online, setup in loaded_setups:
|
||||
- announce = cancel
|
||||
|
||||
nicos == idle, setup not in loaded_setups, announce == cancel:
|
||||
- announce = None, plugplay = cancel
|
||||
|
||||
setup not in loaded_setups, plugplay == cancel:
|
||||
- plugplay = quiet (handled in _handle_setups)
|
||||
|
||||
setup not in loaded_setups, plugplay == quiet and setupfile exists:
|
||||
- remove setupfile and wrapper file (if present)
|
||||
|
||||
"""
|
||||
import time
|
||||
import socket
|
||||
import re
|
||||
from ast import literal_eval
|
||||
from pathlib import Path
|
||||
from frappy.lib import mkthread, formatExtendedTraceback
|
||||
from frappy.core import Writable, Parameter, Property, Command, StringIO, HasIO, nopoll
|
||||
from frappy.core import Readable, Parameter, Property, Command, Communicator, HasIO
|
||||
from frappy.datatypes import ArrayOf, StructOf, StringType, IntRange, TupleOf, BoolType
|
||||
from frappy.client import SecopClient
|
||||
from .marche_frappy import FrappyMarche
|
||||
@@ -33,12 +73,13 @@ from .normalizeuri import normalizeuri
|
||||
porttype = IntRange(0, 0xc000)
|
||||
secnodetype = StructOf(cfg=StringType(), uri=StringType(), status=StringType())
|
||||
|
||||
|
||||
SETUP_TEMPLATE = """description = 'frappy %(cfg)s setup'
|
||||
group = 'plugplay'
|
||||
|
||||
devices = {
|
||||
%(devname)s:
|
||||
device('nicos_sinq.frappy_sinq.devices.FrappyMarcheNode',
|
||||
device('nicos_sinq.frappy_sinq.new.FrappyMarcheNode',
|
||||
description='%(cfg)s SEC node', unit='', async_only=True,
|
||||
prefix='se_', auto_create=True,
|
||||
uri=%(uri)s,
|
||||
@@ -68,34 +109,31 @@ class SecNode:
|
||||
self.client = SecopClient(host_port)
|
||||
self.nodename = cfg
|
||||
self.cfg = cfg
|
||||
self.setup = f'se_{cfg}'
|
||||
self.description = cfg
|
||||
self.online = False
|
||||
self.status = 'created'
|
||||
self.announce = True
|
||||
self._status = 'created'
|
||||
self.trigger = True
|
||||
self.announce_pnp = True # None: do not announce, True: announce new, False: announce removal
|
||||
self.setup_was_loaded = False
|
||||
|
||||
def connect(self, complete_callback=None, log=None):
|
||||
self.status = 'connecting'
|
||||
self._status = 'connecting'
|
||||
self.log = log
|
||||
self.complete_callback = complete_callback
|
||||
# if 'sync':
|
||||
# self.client.disconnect_time = time.time()
|
||||
# self.log.info('connect %r', self.cfg)
|
||||
# self.client._reconnect()
|
||||
# self.log.info('connected %r', self.cfg)
|
||||
# self.complete()
|
||||
# else:
|
||||
if self.log:
|
||||
self.log.info('spawn connect')
|
||||
self.log.debug('spawn connect')
|
||||
self.client.spawn_connect(self.complete)
|
||||
|
||||
def get_setup(self):
|
||||
return f'se_{self.cfg}'
|
||||
|
||||
def complete(self):
|
||||
try:
|
||||
self.online = True
|
||||
self.status = 'completing'
|
||||
self._status = 'completing'
|
||||
self.nodename = self.client.nodename
|
||||
if self.log:
|
||||
self.log.info('connected to %r', self.nodename)
|
||||
self.log.debug('connected to %r', self.nodename)
|
||||
if self.complete_callback:
|
||||
if not self.cfg:
|
||||
if self.nodename == self.client.uri:
|
||||
@@ -109,166 +147,187 @@ class SecNode:
|
||||
self.complete_callback = None
|
||||
desc = self.client.properties.get('description') or self.nodename
|
||||
self.description = desc.split('\n')[0]
|
||||
self.status = 'connected'
|
||||
self._status = 'connected'
|
||||
except Exception as e:
|
||||
self.log.exception('connect failed')
|
||||
self.status = f'disconnected {e!r}'
|
||||
self._status = f'disconnected {e!r}'
|
||||
|
||||
def disconnect(self):
|
||||
self.online = False
|
||||
self.status = 'disconnecting'
|
||||
self._status = 'disconnecting'
|
||||
self.client.disconnect()
|
||||
self.status = 'disconnected'
|
||||
self._status = 'disconnected'
|
||||
|
||||
|
||||
class NicosCacheIO(StringIO):
|
||||
identification = [('nicos/session/mastersetup?', r"nicos/session/mastersetup=\['.*")]
|
||||
def status(self):
|
||||
return self._status
|
||||
|
||||
|
||||
def noop(*args):
|
||||
pass
|
||||
|
||||
|
||||
MSGPAT = re.compile(r'([^=!]*)(?:=(.*)|!)')
|
||||
MSGPAT = re.compile(r'([^=!]*)(?:([=!])(.*))?')
|
||||
|
||||
|
||||
def noop(*args):
|
||||
pass
|
||||
|
||||
|
||||
class NicosCache:
|
||||
def __init__(self, port=14869, host='localhost'):
|
||||
self.host_port = (host, int(port))
|
||||
self.sendqueue = []
|
||||
self.scripts = ''
|
||||
def get_lookup_key(key):
|
||||
split = key.split('/')
|
||||
return f'/{split[-2]}/{split[-1]}' if len(split) > 2 else '/'.join(split[-2:])
|
||||
|
||||
def send(self, key, op, value=''):
|
||||
if op in ('=', '-'):
|
||||
value = repr(value)
|
||||
print('>', key + op + value)
|
||||
self.sock.sendall((key + op + value).encode() + b'\n')
|
||||
|
||||
def purge(self):
|
||||
"""purge pending messages registered"""
|
||||
self.sendqueue.clear()
|
||||
class NicosCache(Communicator):
|
||||
uri = Property('<host>:<port>>', StringType(), default='localhost')
|
||||
_error = None
|
||||
_sock = None
|
||||
|
||||
def run(self, handlers, other=noop, exception=noop):
|
||||
def doPoll(self):
|
||||
self.log.info('doPoll')
|
||||
if self._error:
|
||||
self.log.error('%r', self._error)
|
||||
self._error = None
|
||||
|
||||
@Command(argument=StringType())
|
||||
def communicate(self, command):
|
||||
"""send a command, do not wait for any response"""
|
||||
self._connect()
|
||||
self.log.info('> %r', command)
|
||||
self._sock.sendall(command.encode() + b'\n')
|
||||
|
||||
def _connect(self, keys=()):
|
||||
if self._sock is None:
|
||||
host, _, port = self.uri.partition(':')
|
||||
self._sock = socket.create_connection((host or 'localhost', int(port or 14869)))
|
||||
for lookup_key in keys:
|
||||
msg = f'{lookup_key}*\n{lookup_key}:\n'
|
||||
self._sock.sendall(msg.encode())
|
||||
|
||||
def recvloop(self, handlers, other=noop, exception=noop):
|
||||
handler_lookup = {}
|
||||
for key, handler in handlers.items():
|
||||
pat = re.compile(key.replace('*', '.*'))
|
||||
lookup_key = get_lookup_key(key)
|
||||
handler_lookup.setdefault(lookup_key, []).append((pat, handler))
|
||||
while True:
|
||||
try:
|
||||
self.sock = socket.create_connection(self.host_port)
|
||||
handler_lookup = {} # dict <last two parts of key> of list of (<pat>, <handler>)
|
||||
for key, handler in handlers.items():
|
||||
pat = re.compile(key.replace('*', '.*'))
|
||||
lookup_key = '/%s/%s' % tuple(key.split('/')[-2:])
|
||||
print(f'{lookup_key}*\n{lookup_key}:\n')
|
||||
self.sock.sendall(f'{lookup_key}*\n{lookup_key}:\n'.encode())
|
||||
handler_lookup.setdefault(lookup_key, []).append((pat, handler))
|
||||
self._connect(handler_lookup)
|
||||
self._error = 'connected'
|
||||
except Exception:
|
||||
print(formatExtendedTraceback())
|
||||
self._error = formatExtendedTraceback()
|
||||
time.sleep(10)
|
||||
continue
|
||||
try:
|
||||
buffer = b''
|
||||
while True:
|
||||
raw = self.sock.recv(8192)
|
||||
raw = self._sock.recv(8192)
|
||||
if not raw:
|
||||
break
|
||||
messages = (buffer + raw).split(b'\n')
|
||||
buffer = messages.pop()
|
||||
for msg in messages:
|
||||
msg = msg.strip().decode()
|
||||
print(msg)
|
||||
match = MSGPAT.match(msg)
|
||||
if not match:
|
||||
other(msg)
|
||||
continue
|
||||
key, value = match.groups()
|
||||
lookup_key = '/%s/%s' % tuple(key.split('/')[-2:])
|
||||
# print('KEY=', lookup_key, key, list(handler_lookup.get(lookup_key, ())))
|
||||
key, op, value = match.groups()
|
||||
lookup_key = get_lookup_key(key)
|
||||
for pat, handler in handler_lookup.get(lookup_key, ()):
|
||||
if pat.match(key):
|
||||
print('CALL', pat, handler)
|
||||
try:
|
||||
if value is not None:
|
||||
value = literal_eval(value)
|
||||
handler(key, value)
|
||||
handler(key, op, value)
|
||||
except Exception as e:
|
||||
print(formatExtendedTraceback())
|
||||
self._error = formatExtendedTraceback()
|
||||
exception(e)
|
||||
break
|
||||
else:
|
||||
other(key)
|
||||
messages.extend(raw.split(b'\n'))
|
||||
except Exception as e:
|
||||
print(formatExtendedTraceback())
|
||||
self.sock = None
|
||||
self._error = formatExtendedTraceback()
|
||||
self._sock = None
|
||||
exception(e)
|
||||
raise
|
||||
|
||||
|
||||
class SuperFrappy(HasIO, Writable):
|
||||
ioClass = NicosCacheIO
|
||||
value = Parameter('running servers', ArrayOf(secnodetype), default=())
|
||||
target = Parameter('servers to connect/start', ArrayOf(secnodetype), default=())
|
||||
wrapperdir = Property('frappy wrapper config directory', StringType())
|
||||
setupdir = Property('frappy wrapper config directory', StringType())
|
||||
cfgdirs = Property('frappy config directories to look for', StringType())
|
||||
class SuperFrappy(HasIO, Readable):
|
||||
ioClass = NicosCache
|
||||
marcheport = Property('marche port number', porttype, default=8124)
|
||||
instrument = Property('instrument', StringType())
|
||||
main_port = Property('port number of main server', porttype)
|
||||
stick_port = Property('port number of stick server', porttype)
|
||||
is_main_instrument = Property('this is the main instrument', BoolType(), default=True)
|
||||
value = Parameter('running servers', ArrayOf(secnodetype), default=())
|
||||
instance = Parameter('"main" or <instrument>', StringType())
|
||||
nicos_setups = Parameter('active nicos se setups', ArrayOf(StringType()))
|
||||
_marche = None
|
||||
_secnodes = None # dict <host_post> of SecNode
|
||||
_udp_listener = None
|
||||
_nicos_idle_since = None
|
||||
_setups = ()
|
||||
_to_close = ()
|
||||
|
||||
def initModule(self):
|
||||
super().initModule()
|
||||
if self._marche is None:
|
||||
self._marche = FrappyMarche(self, self.instrument, self.marcheport)
|
||||
self._marche = FrappyMarche(self.instance, 'localhost', self.marcheport)
|
||||
self.setupdir = self._marche.config['setupdir']
|
||||
self.wrapperdir = self._marche.wrapperdir
|
||||
self._secnodes = {}
|
||||
self._to_cleanup = set()
|
||||
self._udp_listener = [UdpScan(True), UdpListener(True)]
|
||||
self.rescan()
|
||||
self._nicos_cache = NicosCache()
|
||||
self._announced_setups = {}
|
||||
self._current_plugplay = {}
|
||||
mkthread(self._nicos_cache.run, {
|
||||
'nicos/session/mastersetupexplicit': self._change_setups,
|
||||
self._to_close = set()
|
||||
# self.log.info('%r', self.log.handlers[0].setLevel(10))
|
||||
mkthread(self.io.recvloop, {
|
||||
'nicos/session/mastersetupexplicit': self._handle_setups,
|
||||
'nicos/exp/scripts': self._handle_scripts,
|
||||
'se/*/nicos/setupname': self._set_plugplay,
|
||||
'se/*/nicos/setupname': self._handle_plugplay,
|
||||
})
|
||||
|
||||
def _handle_scripts(self, key, value):
|
||||
self.log.info('scripts %r', value)
|
||||
def _cache_send(self, key, op, value=''):
|
||||
if op in ('=', '!'):
|
||||
value = repr(value)
|
||||
self.io.communicate(key + op + value)
|
||||
|
||||
def _handle_scripts(self, key, op, value):
|
||||
self.log.debug('scripts %r', value)
|
||||
if value:
|
||||
self.log.info('BUSY')
|
||||
self._nicos_idle_since = None
|
||||
self.setFastPoll(False)
|
||||
else:
|
||||
if not self._nicos_idle_since:
|
||||
self._nicos_idle_since = time.time()
|
||||
self.log.info('TIME %r', self._nicos_idle_since)
|
||||
self.setFastPoll(True, 0.25)
|
||||
|
||||
def _change_setups(self, key, value):
|
||||
def _handle_setups(self, key, op, value):
|
||||
self._setups = set(value)
|
||||
self.log.info('SETUPS %r', value)
|
||||
self.log.debug('setups %r', value)
|
||||
for secnode in self._secnodes.values():
|
||||
if secnode.setup in self._setups and secnode.announce:
|
||||
# the setup was loaded after pnp event -> remove announcement
|
||||
self.log.info('nicos setup %r was loaded', secnode.setup)
|
||||
self._nicos_cache.send(f'se/{secnode.nodename}/nicos/setupname', '=', '')
|
||||
secnode.announce = False
|
||||
setup = secnode.get_setup()
|
||||
loaded = setup in self._setups
|
||||
pnp = self._current_plugplay.get(secnode.nodename)
|
||||
if loaded:
|
||||
if pnp is True:
|
||||
self._send_pnp_message(secnode.nodename, None, True)
|
||||
secnode.setup_was_loaded = True
|
||||
else:
|
||||
if pnp is False:
|
||||
self._send_pnp_message(secnode.nodename, None, False)
|
||||
if secnode.setup_was_loaded:
|
||||
self._to_close.add(secnode.host_port)
|
||||
|
||||
def _set_plugplay(self, key, value):
|
||||
self._current_plugplay[key] = value
|
||||
def _handle_plugplay(self, key, op, value):
|
||||
value = None if value is None else (op == '=')
|
||||
if value is None:
|
||||
self._current_plugplay.pop(key, None)
|
||||
else:
|
||||
self._current_plugplay[key] = value
|
||||
self.log.debug('pnp %r', self._current_plugplay)
|
||||
|
||||
def _write_setup_file(self, secnode):
|
||||
setup_file = Path(self.setupdir) / f'{secnode.setup}.py'
|
||||
setup_file = Path(self.setupdir) / f'{secnode.get_setup()}.py'
|
||||
envlist, alias_config, devmap = self.node_setup_info(secnode)
|
||||
nodeargs = f'device_mapping={devmap!r}' if devmap else ''
|
||||
aliasconfig = f'alias_config = {alias_config!r}' if alias_config else ''
|
||||
@@ -277,29 +336,53 @@ class SuperFrappy(HasIO, Writable):
|
||||
'uri': repr(secnode.host_port), 'nodeargs': nodeargs, 'aliasconfig': aliasconfig}
|
||||
setup_file.write_text(setup_content)
|
||||
|
||||
def _update_plugplay(self):
|
||||
def _send_pnp_message(self, nodename, setup, on):
|
||||
self._current_plugplay[nodename] = None if setup is None else on
|
||||
self._cache_send(f'se/{nodename}/nicos/setupname', '=' if on else '!', setup)
|
||||
|
||||
def _update(self):
|
||||
while self._to_close:
|
||||
secnode = self._secnodes.pop(self._to_close.pop(), None)
|
||||
self._remove_secnode(secnode)
|
||||
superfluous_setup_files = set(Path(self.setupdir).glob('*.py'))
|
||||
superfluous_cfg_files = set(Path(self.wrapperdir).glob('*_cfg.py'))
|
||||
for secnode in self._secnodes.values():
|
||||
nodename = secnode.nodename.replace('/', '_')
|
||||
setup_file = Path(self.setupdir) / f'{secnode.setup}.py'
|
||||
if not secnode.cfg:
|
||||
continue
|
||||
setup = secnode.get_setup()
|
||||
setup_file = Path(self.setupdir) / f'{setup}.py'
|
||||
superfluous_setup_files.discard(setup_file)
|
||||
if secnode.online:
|
||||
if self._current_plugplay.get(nodename) != secnode.setup:
|
||||
self._write_setup_file(secnode)
|
||||
if secnode.announce:
|
||||
self.log.info('announce %s', secnode.setup)
|
||||
self._nicos_cache.send(f'se/{nodename}/nicos/setupname', '=', secnode.setup)
|
||||
self._current_plugplay[nodename] = secnode.setup
|
||||
superfluous_cfg_files.discard(Path(self.wrapperdir) / f'{secnode.cfg}_cfg.py')
|
||||
if setup in self._setups:
|
||||
if not secnode.online:
|
||||
if secnode.announce_pnp is False:
|
||||
if self._nicos_idle_since:
|
||||
secnode.announce_pnp = None
|
||||
self._send_pnp_message(nodename, setup, False)
|
||||
elif secnode.host_port not in self._to_close:
|
||||
secnode.announce_pnp = False
|
||||
else:
|
||||
setupname = self._current_plugplay.pop(nodename, None)
|
||||
if setupname:
|
||||
self._nicos_cache.send(f'se/{nodename}/nicos/setupname', '!', '')
|
||||
for file in superfluous_setup_files:
|
||||
self.log.info('remove %s', file)
|
||||
if secnode.online:
|
||||
if not setup_file.is_file():
|
||||
self.log.debug('write_setup %r', secnode.cfg)
|
||||
self._write_setup_file(secnode)
|
||||
if secnode.announce_pnp is True and self._nicos_idle_since:
|
||||
secnode.announce_pnp = None
|
||||
self._send_pnp_message(nodename, setup, True)
|
||||
self._to_close.discard(secnode.host_port)
|
||||
for setup in self._setups:
|
||||
# do not delete active setups
|
||||
superfluous_setup_files.discard(Path(self.setupdir) / f'{setup}.py')
|
||||
reload_marche = bool(superfluous_cfg_files)
|
||||
for file in superfluous_setup_files | superfluous_cfg_files:
|
||||
self.log.debug('remove %s', file)
|
||||
try:
|
||||
file.unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
if reload_marche:
|
||||
self._marche.reload()
|
||||
|
||||
def doPoll(self):
|
||||
super().doPoll()
|
||||
@@ -308,16 +391,16 @@ class SuperFrappy(HasIO, Writable):
|
||||
if msg:
|
||||
uri = msg['uri']
|
||||
if uri in self._secnodes:
|
||||
self.log.info('%r is already known', msg)
|
||||
self.log.debug('%r is already known', msg)
|
||||
else:
|
||||
self.log.info('%r appeared', msg)
|
||||
self.log.debug('%r appeared', msg)
|
||||
cfg = msg.get('device', '')
|
||||
if uri.startswith('localhost:') and cfg != 'superfrappy':
|
||||
self.connect(uri)
|
||||
now = time.time()
|
||||
if now > (self._nicos_idle_since or now) + 0.5:
|
||||
self.setFastPoll(False)
|
||||
self._update_plugplay()
|
||||
self._update()
|
||||
|
||||
@Command
|
||||
def rescan(self):
|
||||
@@ -327,45 +410,20 @@ class SuperFrappy(HasIO, Writable):
|
||||
|
||||
def read_value(self):
|
||||
value = []
|
||||
target = []
|
||||
for secnode in self._secnodes.values():
|
||||
info = {'cfg': secnode.cfg, 'uri': secnode.host_port, 'status': secnode.status}
|
||||
value.append(info) # TODO: check for online
|
||||
if not secnode.status.startswith('disconn'):
|
||||
target.append(info)
|
||||
self.target = target
|
||||
info = {'cfg': secnode.cfg, 'uri': secnode.host_port, 'status': secnode.status()}
|
||||
value.append(info)
|
||||
return value
|
||||
|
||||
@nopoll
|
||||
def read_target(self):
|
||||
self.read_value()
|
||||
|
||||
def write_target(self, target):
|
||||
pass # TODO: go through add_secnode / remove_secnode
|
||||
|
||||
def cleanup(self):
|
||||
marche_status = self._marche.status('frappy')
|
||||
for secnode in list(self._to_cleanup):
|
||||
if secnode.host_port not in self._secnodes:
|
||||
self._to_cleanup.discard(secnode)
|
||||
self.log.info('cleanup (node has gone) %r', self._to_cleanup)
|
||||
continue
|
||||
busy, running, _ = marche_status.get(secnode.cfg, (False, False, ''))
|
||||
if busy or running:
|
||||
continue
|
||||
if secnode.cfg not in self._setups: # setup is unloaded
|
||||
try:
|
||||
(Path(self.setupdir) / f'{secnode.cfg}.py').unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
self._to_cleanup.discard(secnode)
|
||||
self.log.info('cleanup (setup in nicos has gone) %r', self._to_cleanup)
|
||||
self._secnodes.pop(secnode.host_port, None)
|
||||
|
||||
def read_nicos_setups(self):
|
||||
return list(self._setups & {s.setup for s in self._secnodes.values()})
|
||||
return list(self._setups & {s.get_setup() for s in self._secnodes.values()})
|
||||
|
||||
@Command()
|
||||
def ping(self):
|
||||
"""nicos cache ping"""
|
||||
self._cache_send('#ping#', '?', '')
|
||||
|
||||
@Command(argument=StringType())
|
||||
def restart(self, cfg):
|
||||
"""restart the frappy server
|
||||
|
||||
@@ -384,16 +442,30 @@ class SuperFrappy(HasIO, Writable):
|
||||
host_port = normalizeuri(host_port, True)
|
||||
secnode = self._secnodes.get(host_port)
|
||||
if secnode:
|
||||
self.log.info('already connected %r', host_port)
|
||||
self.log.debug('already connected %r', host_port)
|
||||
else:
|
||||
self.log.info('connect secnode %r', host_port)
|
||||
self.log.debug('connect secnode %r', host_port)
|
||||
secnode = SecNode(host_port, self._marche.cfg_info.get(host_port, ''))
|
||||
self._secnodes[host_port] = secnode
|
||||
secnode.connect(self._add_setup, self.log)
|
||||
secnode.connect(log=self.log)
|
||||
self.read_value()
|
||||
|
||||
@Command(argument=TupleOf(StringType(), StringType()))
|
||||
def add(self, service, cfg):
|
||||
@Command(argument=StringType())
|
||||
def main(self, cfg):
|
||||
"""add or change main cfg"""
|
||||
self._add_secnode('main', cfg)
|
||||
|
||||
@Command(argument=StringType())
|
||||
def stick(self, cfg):
|
||||
"""add or change stick cfg"""
|
||||
self._add_secnode('stick', cfg)
|
||||
|
||||
@Command(argument=StringType())
|
||||
def add(self, cfg):
|
||||
"""add an addons cfg"""
|
||||
self._add_secnode('addons', cfg)
|
||||
|
||||
def _add_secnode(self, service, cfg):
|
||||
"""add and start server on localhost
|
||||
|
||||
- add a wrapper cfg file to the wrapper dir
|
||||
@@ -402,57 +474,25 @@ class SuperFrappy(HasIO, Writable):
|
||||
:param service: 'stick', 'main', '' or a stringified port number
|
||||
:param cfg: config file or equipment id
|
||||
"""
|
||||
self.log.info('add and start %r', cfg)
|
||||
if service == 'main':
|
||||
port = self.main_port
|
||||
elif service == 'stick':
|
||||
port = self.stick_port
|
||||
elif service:
|
||||
port = int(service)
|
||||
service = 'addons'
|
||||
else:
|
||||
service = 'addons'
|
||||
used_ports = set()
|
||||
for secnode in self._secnodes.values():
|
||||
if secnode.cfg == cfg:
|
||||
self.log.info('%s is already added at %s', cfg, secnode.host_port)
|
||||
return
|
||||
else:
|
||||
used_ports.add(int(secnode.port))
|
||||
else:
|
||||
incr = self.stick_port - self.main_port
|
||||
portnr = self.stick_port + incr
|
||||
while portnr in used_ports:
|
||||
portnr += incr
|
||||
port = portnr
|
||||
self.log.debug('add and start %r', cfg)
|
||||
port = self._marche.get_port(service)
|
||||
host_port = f'localhost:{port}'
|
||||
secnode = SecNode(host_port, cfg)
|
||||
self._secnodes[host_port] = secnode
|
||||
self._marche.add_frappy_service(service, secnode.cfg, secnode.port, self.log)
|
||||
self.log.info('start %r at %r', secnode.cfg, host_port)
|
||||
self.log.debug('start %r at %r', secnode.cfg, host_port)
|
||||
self._marche.start(secnode.cfg)
|
||||
secnode.connect(self._add_setup, self.log)
|
||||
secnode.connect(log=self.log)
|
||||
self.read_value()
|
||||
|
||||
def _add_setup(self, secnode):
|
||||
envlist, alias_config, devmap = self.node_setup_info(secnode)
|
||||
nodeargs = f'device_mapping={devmap!r}' if devmap else ''
|
||||
aliasconfig = f'alias_config = {alias_config!r}' if alias_config else ''
|
||||
setup_content = SETUP_TEMPLATE % {
|
||||
'cfg': secnode.cfg, 'devname': f'"secnode_{secnode.cfg}"',
|
||||
'uri': repr(secnode.host_port), 'nodeargs': nodeargs, 'aliasconfig': aliasconfig}
|
||||
(Path(self.setupdir) / f'{secnode.cfg}.py').write_text(setup_content)
|
||||
nodename = secnode.nodename.replace('/', '_')
|
||||
# announce plugplay to nicos
|
||||
self._nicos_cache.send(nodename, f'se/{nodename}/nicos/description={secnode.description!r}')
|
||||
self._nicos_cache.send(nodename, f'se/{nodename}/nicos/setupname={secnode.cfg!r}?')
|
||||
self.log.info('added server %s %s', secnode.cfg, secnode.host_port)
|
||||
|
||||
def _reset_setup(self, secnode):
|
||||
nodename = secnode.nodename.replace('/', '_')
|
||||
# announce plugplay removal to nicos
|
||||
self.communicate(f"se/{nodename}/nicos/setupname!{secnode.cfg!r}\nse/{nodename}/nicos/setupname?")
|
||||
self.log.info('reset server %s %s', secnode.cfg, secnode.host_port)
|
||||
def _remove_secnode(self, secnode):
|
||||
secnode.announce = False
|
||||
secnode.disconnect()
|
||||
self.log.debug('secnode.host %r', secnode)
|
||||
if secnode.host == 'localhost':
|
||||
self._marche.stop(secnode.cfg)
|
||||
self._update()
|
||||
self.read_value()
|
||||
|
||||
@Command(argument=StringType())
|
||||
def remove(self, host_port_or_cfg):
|
||||
@@ -468,14 +508,8 @@ class SuperFrappy(HasIO, Writable):
|
||||
host_port = normalizeuri(host_port_or_cfg, True)
|
||||
secnode = self._secnodes.get(host_port)
|
||||
if secnode:
|
||||
# TODO: announce removal of setup to plugplay
|
||||
secnode.disconnect()
|
||||
self._to_cleanup.add(secnode)
|
||||
self._reset_setup(secnode)
|
||||
self.log.info('to cleanup %r', self._to_cleanup)
|
||||
self._marche.delete_frappy_service(secnode.cfg)
|
||||
if secnode.host == 'localhost':
|
||||
self._marche.stop(secnode.cfg)
|
||||
self._remove_secnode(secnode)
|
||||
self._update()
|
||||
self.read_value()
|
||||
else:
|
||||
raise ValueError(f'no running server found for {host_port_or_cfg}')
|
||||
|
||||
@@ -20,13 +20,26 @@
|
||||
import sys
|
||||
import time
|
||||
import re
|
||||
import socket
|
||||
from pathlib import Path
|
||||
from configparser import ConfigParser
|
||||
import logging
|
||||
|
||||
|
||||
MARCHESRC = ['/home/software/marche']
|
||||
CFGDIRS = ['/home/linse/config', '/home/l_samenv/linse_config']
|
||||
|
||||
|
||||
def get_logger(previous=[]):
|
||||
if previous:
|
||||
return previous[0]
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(logging.INFO)
|
||||
logger.addHandler(logging.StreamHandler(sys.stdout))
|
||||
previous.append(logger)
|
||||
return logger
|
||||
|
||||
|
||||
for marchedir in MARCHESRC:
|
||||
if Path(marchedir).is_dir():
|
||||
sys.path.append(marchedir)
|
||||
@@ -65,15 +78,17 @@ class MarcheControl:
|
||||
|
||||
def __init__(self, host, port=None, user=None, instrument=None):
|
||||
self.host = host
|
||||
self.instrument = instrument or host
|
||||
self.user = user or host # SINQ instruments
|
||||
self.instrument = instrument or socket.gethostname().split('.')[0]
|
||||
self.user = user or instrument # SINQ instruments
|
||||
if port is not None:
|
||||
self.port = port
|
||||
self._client = None
|
||||
|
||||
def connect(self):
|
||||
if self._client is None:
|
||||
self._client = Client(self.host, self.port, self.user, self.host.upper() + 'LNS')
|
||||
# TODO: may need more generic solution for last arg
|
||||
print(self.host, self.port, self.user)
|
||||
self._client = Client(self.host, self.port, self.user, self.instrument.upper() + 'LNS')
|
||||
|
||||
# TODO; do we need disconnect?
|
||||
|
||||
@@ -82,6 +97,7 @@ class MarcheControl:
|
||||
|
||||
def start(self, instance):
|
||||
self.connect()
|
||||
print(self.get_service(instance))
|
||||
self._client.startService(self.get_service(instance))
|
||||
|
||||
def restart(self, instance):
|
||||
@@ -95,33 +111,68 @@ class MarcheControl:
|
||||
def status(self, service):
|
||||
"""returns a dict <service> of (<busy>, <running (now or soon)>, <state name>)"""
|
||||
self.connect()
|
||||
statedict = self._client.getAllServiceInfo()[
|
||||
service]['instances']
|
||||
servdict = self._client.getAllServiceInfo().get(service)
|
||||
if not servdict:
|
||||
return {}
|
||||
statedict = servdict['instances']
|
||||
return {k: STATUS_MAP[v['state']] for k, v in statedict.items()}
|
||||
|
||||
def reload(self):
|
||||
self.connect()
|
||||
self._client.reloadJobs()
|
||||
|
||||
def run(self, action, instance=None, *args):
|
||||
if action == 'start':
|
||||
self.start(instance)
|
||||
elif action == 'restart':
|
||||
self.restart(instance)
|
||||
elif action == 'stop':
|
||||
self.stop(instance)
|
||||
else:
|
||||
raise ValueError('unknown args %r', (action, instance) + args)
|
||||
|
||||
WRAPPER_CFG = """include({cfg!r})
|
||||
overrideNode(interface={port!r})
|
||||
|
||||
WRAPPER_CFG = """interface = '{port}'
|
||||
include({cfg!r})
|
||||
overrideNode(interface=interface)
|
||||
"""
|
||||
WRAPPER_PAT = re.compile(r'.*interface=["\'](\d*)["\']', re.DOTALL)
|
||||
WRAPPER_PAT = re.compile(r"interface\s=\s*'(\d*)'\s*\n")
|
||||
|
||||
|
||||
class FrappyMarche(MarcheControl):
|
||||
|
||||
def __init__(self, config, host, port=None, user=None, instrument=None):
|
||||
self.config = config
|
||||
def __init__(self, instance, host='localhost', port=None, user=None):
|
||||
parser = ConfigParser()
|
||||
parser.optionxform = str
|
||||
gencfg = f'~/.config/frappy/{instance}.cfg'
|
||||
parser.read([str(Path(gencfg).expanduser())])
|
||||
try:
|
||||
section = dict(parser['superfrappy'])
|
||||
except KeyError:
|
||||
raise ValueError(f'bad config {gencfg}')
|
||||
self.instance = instance # 'main' or an instrument on a generic computer
|
||||
instrument = section.get('instrument', instance)
|
||||
if instrument == 'main':
|
||||
instrument = socket.gethostname().split('.')[0]
|
||||
print(instance, instrument)
|
||||
self.instrument = instrument # the instrument name
|
||||
|
||||
self.config = {k: section[k].replace('<INS>', instrument) for k in section}
|
||||
|
||||
self.wrapperdir = self.config.pop('wrapperdir')
|
||||
self.cfgdirs = self.config.pop('cfgdirs')
|
||||
self.main_port = int(self.config.pop('main_port'))
|
||||
|
||||
if not Path(self.wrapperdir).is_dir():
|
||||
raise ValueError(f'{self.wrapperdir} does not exist')
|
||||
self.get_cfg_info() # do we need to update this from time to time?
|
||||
super().__init__(host, port, user, instrument)
|
||||
|
||||
def get_service(self, instance):
|
||||
return f'frappy.{instance}' if self.config.is_main_instrument else f'frappy.{self.instrument}-{instance}'
|
||||
return f'frappy.{instance}' if self.instance == 'main' else f'frappy.{self.instrument}-{instance}'
|
||||
|
||||
def wrapper_file(self, cfg):
|
||||
return Path(self.config.wrapperdir) / f'{cfg}_cfg.py'
|
||||
return Path(self.wrapperdir) / f'{cfg}_cfg.py'
|
||||
|
||||
def cfg_file(self, cfgdirs, service, cfg):
|
||||
cfgpy = f'{cfg}_cfg.py'
|
||||
@@ -135,22 +186,56 @@ class FrappyMarche(MarcheControl):
|
||||
else:
|
||||
raise FileNotFoundError(f'can not find {cfgpy} in {tries}')
|
||||
|
||||
def get_std_port(self, service):
|
||||
port = self.main_port
|
||||
if service == 'main':
|
||||
return port
|
||||
port += 1
|
||||
if service == 'stick':
|
||||
return port
|
||||
return port + 1, self.main_port + 10
|
||||
|
||||
def get_local_ports(self):
|
||||
self.get_cfg_info()
|
||||
run_state = self.status('frappy')
|
||||
result = {}
|
||||
for host_port, cfg in self.cfg_info.items():
|
||||
host, port = host_port.split(':')
|
||||
if host == 'localhost':
|
||||
busy, on, _ = run_state.get(cfg, (0,0,0))
|
||||
if on or busy:
|
||||
result.setdefault(port, []).append((on, busy, cfg))
|
||||
return {sorted(v)[-1][-1]: k for k, v in result.items()}
|
||||
|
||||
def get_port(self, service):
|
||||
if service not in {'main', 'stick', 'addons', 'addon'}:
|
||||
raise ValueError('illegal service argument')
|
||||
ports = self.get_std_port(service)
|
||||
if isinstance(ports, int):
|
||||
return ports
|
||||
self.get_cfg_info()
|
||||
used_ports = self.get_local_ports()
|
||||
for port in range(*ports):
|
||||
if port not in used_ports:
|
||||
return port
|
||||
raise ValueError('too many frappy servers')
|
||||
|
||||
def add_frappy_service(self, service, cfg, port, log=None):
|
||||
if log:
|
||||
log.info('add %r port=%r', cfg, port)
|
||||
cfgfile = self.cfg_file(self.config.cfgdirs, service, cfg)
|
||||
if log is None:
|
||||
log = get_logger()
|
||||
log.info('add %r port=%r', cfg, port)
|
||||
cfgfile = self.cfg_file(self.cfgdirs, service, cfg)
|
||||
wrapper_content = WRAPPER_CFG.format(cfg=str(cfgfile), port=port)
|
||||
self.wrapper_file(cfg).write_text(wrapper_content)
|
||||
self.get_cfg_info()
|
||||
if log:
|
||||
log.info('wrapper %r %r', self.wrapper_file(cfg), wrapper_content)
|
||||
log.info('wrapper %r %r', self.wrapper_file(cfg), wrapper_content)
|
||||
self.reload()
|
||||
log.info('registered %r', cfg)
|
||||
|
||||
def get_cfg_info(self):
|
||||
"""get info from wrapper dir"""
|
||||
result = {}
|
||||
for cfgfile in Path(self.config.wrapperdir).glob('*_cfg.py'):
|
||||
for cfgfile in Path(self.wrapperdir).glob('*_cfg.py'):
|
||||
cfg = cfgfile.stem[:-4]
|
||||
match = WRAPPER_PAT.match(cfgfile.read_text())
|
||||
if match:
|
||||
@@ -163,3 +248,24 @@ class FrappyMarche(MarcheControl):
|
||||
self.reload()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
def run(self, action, *args):
|
||||
if action == 'start':
|
||||
try:
|
||||
service, cfg = args
|
||||
except ValueError:
|
||||
raise ValueError('start needs <service> <cfg>')
|
||||
port = self.get_port(service)
|
||||
self.add_frappy_service(service, cfg, port)
|
||||
self.start(cfg)
|
||||
elif action == 'restart':
|
||||
if len(args) != 1 or args[0] in {'main', 'stick', 'addons', 'addon'}:
|
||||
raise ValueError('restart needs <cfg>')
|
||||
self.restart(args[0])
|
||||
elif action == 'stop':
|
||||
if len(args) != 1 or args[0] in {'main', 'stick', 'addons', 'addon'}:
|
||||
raise ValueError('stop needs <cfg>')
|
||||
self.delete_frappy_service(args[0])
|
||||
self.stop(args[0])
|
||||
else:
|
||||
raise ValueError('unknown action %r', action)
|
||||
|
||||
@@ -46,7 +46,7 @@ class Listener:
|
||||
addr = socket.getnameinfo(addr, socket.NI_NOFQDN)[0]
|
||||
msg = json.loads(msg.decode('utf-8'))
|
||||
if log:
|
||||
log.info('got msg %r', msg)
|
||||
log.debug('got msg %r', msg)
|
||||
kind = msg.pop('SECoP', None)
|
||||
if kind == 'node':
|
||||
msg['device'] = msg['equipment_id'].split('.')[0]
|
||||
|
||||
Reference in New Issue
Block a user