diff --git a/cfg/superfrappy_cfg.py b/cfg/superfrappy_cfg.py index 4b6ba65d..6c3d95eb 100644 --- a/cfg/superfrappy_cfg.py +++ b/cfg/superfrappy_cfg.py @@ -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 = [], - ) \ No newline at end of file + ) + diff --git a/frappy_psi/jtccr.py b/frappy_psi/jtccr.py index 05ded912..2c3c3000 100644 --- a/frappy_psi/jtccr.py +++ b/frappy_psi/jtccr.py @@ -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 = { diff --git a/frappy_psi/superfrappy/__init__.py b/frappy_psi/superfrappy/__init__.py index aece2ff4..b6ed4ba0 100644 --- a/frappy_psi/superfrappy/__init__.py +++ b/frappy_psi/superfrappy/__init__.py @@ -16,13 +16,53 @@ # Module authors: # Markus Zolliker # ***************************************************************************** +"""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(':>', 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 of list of (, ) - 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 ', StringType()) nicos_setups = Parameter('active nicos se setups', ArrayOf(StringType())) _marche = None _secnodes = None # dict 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}') diff --git a/frappy_psi/superfrappy/marche_frappy.py b/frappy_psi/superfrappy/marche_frappy.py index f5f53390..3453f031 100644 --- a/frappy_psi/superfrappy/marche_frappy.py +++ b/frappy_psi/superfrappy/marche_frappy.py @@ -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 of (, , )""" 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('', 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 ') + 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 ') + self.restart(args[0]) + elif action == 'stop': + if len(args) != 1 or args[0] in {'main', 'stick', 'addons', 'addon'}: + raise ValueError('stop needs ') + self.delete_frappy_service(args[0]) + self.stop(args[0]) + else: + raise ValueError('unknown action %r', action) diff --git a/frappy_psi/superfrappy/secop_udp.py b/frappy_psi/superfrappy/secop_udp.py index 883e6813..d37c3de2 100644 --- a/frappy_psi/superfrappy/secop_udp.py +++ b/frappy_psi/superfrappy/secop_udp.py @@ -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]