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

added frappy_config to frappy setup

Browse Source 
frappy_config allows to configure how SE aliases are generated
set_se_list applies this to create aliases and update the envrionment
This commit is contained in:
zolliker
2022-04-20 12:06:58 +02:00
parent 80d81d72ae
commit 1d39bd162b
3 changed files with 105 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
commands.py
View File

@ -26,6 +26,7 @@ from glob import glob
from configparser import ConfigParser
from nicos import session, config
from nicos.core import DeviceAlias
from nicos.utils import printTable
from nicos.commands import helparglist, usercommand
from nicos.commands.basic import AddSetup, CreateAllDevices, CreateDevice
@ -68,38 +69,100 @@ def all_info(all_cfg):
def applyAliasConfig():
"""Apply the desired aliases from session.alias_config.
in addition, return the names of targets inheriting from SecopDevice
as a list
be more quiet than original
"""
# reimplemented from Session.applyAliasConfig
# apply also when target dev name does not change, as the target device might have
# be exchanged in the mean time
targets = []
unused = set()
for aliasname, targets in session.alias_config.items():
if aliasname not in session.devices:
# complain about this; setups should make sure that the device
# exists when configuring it
session.log.warning("alias device '%s' does not exist, cannot set"
' its target', aliasname)
continue
continue # silently ignore
aliasdev = session.getDevice(aliasname)
for target, _ in sorted(targets, key=lambda t: -t[1]):
if target in session.devices:
try:
aliasdev.alias = target
targets.append(target.name)
except Exception:
session.log.exception("could not set '%s' alias", aliasdev)
break
def remove_se_aliases():
frappy_config = session.getDevice('frappy_config')
configured = {}
for meaning in frappy_config.meanings:
try:
aliasname, aliascfg = getattr(frappy_config, meaning)
except TypeError:
continue
aliasdev = session.devices.get(aliasname)
if aliasdev:
session.destroyDevice(aliasname)
session.configured_devices.pop(aliasname, None)
session.dynamic_devices.pop(aliasname, None)
@usercommand
def set_se_list():
"""create aliases for SECoP devices
depending on their meaning
"""
remove_se_aliases()
frappy_config = session.getDevice('frappy_config')
nodedevs = filter(None, [session.devices.get(devname) for devname in frappy_config.nodes])
sample_devices = {}
for nodedev in nodedevs:
secnode = nodedev._secnode
if not secnode:
continue
for devname, (_, desc) in nodedev.setup_info.items():
secop_module = desc['secop_module']
meaning = secnode.modules[secop_module]['properties'].get('meaning')
if meaning:
meaning_name, importance = meaning
sample_devices.setdefault(meaning_name, []).append((importance * 10 + 5, devname))
newlist = []
previous = session.experiment.envlist[:]
for meaning in frappy_config.meanings:
try:
aliasname, aliascfg = getattr(frappy_config, meaning)
except TypeError:
continue
importance_list = sample_devices.get(meaning, [])
importance_list.extend([(nr, nam) for nam, nr in aliascfg.items() if nam in session.devices])
importance_list = sorted(importance_list, reverse=True)
session.log.debug('%s: %r', meaning, importance_list)
for _, devname, in importance_list:
dev = session.devices.get(devname)
if dev:
session.log.debug('create alias %r pointing to %r', aliasname, devname)
devcfg = ('nicos.core.DeviceAlias', {})
session.configured_devices[aliasname] = devcfg
session.dynamic_devices[aliasname] = 'frappy'
aliasdev = session.createDevice(aliasname, recreate=True, explicit=True)
aliasdev.alias = devname
if aliasname not in previous:
newlist.append(aliasname)
break
else:
session.log.warning('none of the desired targets for alias %r '
'actually exist', aliasname)
return targets
if aliasname in previous:
previous.remove(aliasname)
applyAliasConfig() # for other aliases
newlist = previous + newlist
if set(newlist) != set(session.experiment.envlist):
session.experiment.setEnvironment(newlist)
session.log.info('changed environment to: %s', ', '.join(session.experiment.envlist))
def frappy_start(service, cfg=None):
if service not in SERVICES:
raise ValueError('unknown service %s' % service)
remove_se_aliases()
if cfg == '':
seaconn = session.devices.get('seaconn')
if seaconn and seaconn._attached_secnode:
@ -138,16 +201,7 @@ def frappy_start(service, cfg=None):
if cfg is not None:
cleanup_defunct()
CreateAllDevices()
targets = applyAliasConfig()
newlist = []
for itemname in session.experiment.envlist:
devname = itemname.split(':')[0]
dev = session.devices.get(devname)
if dev is not None and (not isinstance(dev, SecopDevice) or devname in targets):
newlist.append(itemname)
newlist.extend(t for t in targets if t not in newlist)
if newlist != session.experiment.envlist:
session.experiment.setEnvironment(newlist)
set_se_list()
if startnode and cfg == '':
startnode.disable()
return all_cfg

17
devices.py
View File

@ -33,6 +33,7 @@ from os.path import expanduser
from nicos import config, session
from nicos.core import Override, Param, Moveable, status
from nicos.devices.secop import SecNodeDevice
from nicos.core import Device, anytype, listof
from nicos.utils.comparestrings import compare
from servicemanager import FrappyManager
@ -45,6 +46,22 @@ def suggest(poi, allowed_keys):
return [m[0] for m in comp[:3] if m[1] > 2]
class FrappyConfig(Device):
parameters = {
'temperature': Param(
'device name for sample temperature', type=anytype, default=['Ts', []]),
'temperature_regulation': Param(
'device name for temperature regulation', type=anytype, default=['Tr', []]),
'magneticfield': Param(
'device name for magnetic field', type=anytype, default=['B', []]),
'nodes': Param(
'list of names of potential SEC nodes', type=listof(str), default=[]),
}
meanings = list(parameters)
meanings.remove('nodes')
class FrappyNode(SecNodeDevice, Moveable):
"""SEC node device

20
setups/frappy.py
View File

@ -3,14 +3,17 @@ group = 'optional'
modules = ['nicos_sinq.frappy_sinq.commands']
devices = {
'temperature': device('nicos.devices.generic.DeviceAlias'),
'magfield': device('nicos.devices.generic.DeviceAlias'),
}
alias_config = {
'temperature': {'se_ts': 110, 'se_tt': 100},
'magfield': {'se_mf': 100},
}
devices = dict(
frappy_config = device(
'nicos_sinq.frappy_sinq.devices.FrappyConfig',
visibility = [],
nodes = ['se_main', 'se_stick', 'se_addons'],
# T will be either a SECoP module with property meaning=tmperature or one of the keys in the dict
temperature = ['T', {'se_ts': 120, 'se_tt': 100}],
temperature_regulation = ['Tr', {'se_tm': 110, 'se_tv': 100}],
magneticfield = ['B', {'se_mf': 100}],
)
)
startupcode = '''
printinfo("=======================================================================================")
@ -23,4 +26,5 @@ printinfo(" frappy_stick('') # remove stick")
printinfo(" frappy_main('') # remove main SE apparatus")
printinfo(" frappy_main() # show the current SE configuration")
printinfo("=======================================================================================")
set_se_list()
'''