linse/frappy
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
frappy/secop/client/__init__.py
Markus Zolliker 97034fb998 implement SECoP proxy modules 
A proxy module is a module with a known structure, but
accessed over a SECoP connection.
For the configuration, a Frappy module class has to be given.
The proxy class is created from this, but does not inherit from it.
However, the class of the returned object will be subclass of the
SECoP base classes (Readable, Drivable etc.).
A possible extension might be, that instead of the Frappy class,
the JSON module description can be given, as a separate file
or directly in the config file.
Or we might offer a tool to convert the JSON description to
a python class.

Change-Id: I9212d9f3fe82ec56dfc08611d0e1efc0b0112271
Reviewed-on: https://forge.frm2.tum.de/review/c/sine2020/secop/playground/+/22386
Tested-by: JenkinsCodeReview <bjoern_pedersen@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
2020-02-07 12:46:04 +01:00

539 lines
22 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
# *****************************************************************************
#
# 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
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# 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
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# 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.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
# Module authors:
# Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
# Markus Zolliker <markus.zolliker@psi.ch>
#
# *****************************************************************************
"""general SECoP client"""
import time
import queue
import json
from threading import Event, RLock
from collections import defaultdict
from secop.lib import mkthread, formatExtendedTraceback, formatExtendedStack
from secop.lib.asynconn import AsynConn, ConnectionClosed
from secop.datatypes import get_datatype
from secop.protocol.interface import encode_msg_frame, decode_msg
from secop.protocol.messages import REQUEST2REPLY, ERRORPREFIX, EVENTREPLY, WRITEREQUEST, WRITEREPLY, \
READREQUEST, READREPLY, IDENTREQUEST, IDENTPREFIX, ENABLEEVENTSREQUEST, COMMANDREQUEST, DESCRIPTIONREQUEST
import secop.errors
import secop.params
# replies to be handled for cache
UPDATE_MESSAGES = {EVENTREPLY, READREPLY, WRITEREPLY, ERRORPREFIX + READREQUEST, ERRORPREFIX + EVENTREPLY}
class UNREGISTER:
"""a magic value, used a returned value in a callback
to indicate it has to be unregistered
used to implement one shot callbacks
"""
class Logger:
"""dummy logger, in case not provided from caller"""
@staticmethod
def info(fmt, *args, **kwds):
print(str(fmt) % args)
@staticmethod
def noop(fmt, *args, **kwds):
pass
debug = noop
error = warning = critical = info
class CallbackObject:
"""abstract definition for a target object for callbacks
this is mainly for documentation, but it might be extended
and used as a mixin for objects registered as a callback
"""
def updateEvent(self, module, parameter, value, timestamp, readerror):
"""called whenever a value is changed
or when new callbacks are registered
"""
def unhandledMessage(self, action, ident, data):
"""called on an unhandled message"""
def nodeStateChange(self, online, state):
"""called when the state of the connection changes
'online' is True when connected or reconnecting, False when disconnected or connecting
'state' is the connection state as a string
"""
def descriptiveDataChange(self, module, description):
"""called when the description has changed
this callback is called on the node with module=None
and on every changed module with module==<module name>
"""
class ProxyClient:
"""common functionality for proxy clients"""
CALLBACK_NAMES = ('updateEvent', 'descriptiveDataChange', 'nodeStateChange', 'unhandledMessage')
online = False # connected or reconnecting since a short time
validate_data = False
_state = 'disconnected' # further possible values: 'connecting', 'reconnecting', 'connected'
def __init__(self):
self.callbacks = {cbname: defaultdict(list) for cbname in self.CALLBACK_NAMES}
# caches (module, parameter) = value, timestamp, readerror (internal names!)
self.cache = {}
def register(self, key, obj=None, **kwds):
"""register callback functions
- kwds each key must be a valid callback name defined in self.CALLBACK_NAMES
- kwds values are the callback functions
- if obj is not None, use its methods named from the callback name, if not given in kwds
- key might be either:
1) None: general callback (all callbacks)
2) <module name>: callbacks related to a module (not called for 'unhandledMessage')
3) (<module name>, <parameter name>): callback for specified parameter (only called for 'updateEvent')
"""
for cbname in self.CALLBACK_NAMES:
cbfunc = kwds.pop(cbname, None)
if obj and cbfunc is None:
cbfunc = getattr(obj, cbname, None)
if not cbfunc:
continue
cbdict = self.callbacks[cbname]
cbdict[key].append(cbfunc)
# immediately call for some callback types
if cbname == 'updateEvent':
if key is None:
for (mname, pname), data in self.cache.items():
cbfunc(mname, pname, *data)
else:
data = self.cache.get(key, None)
if data:
cbfunc(*key, *data) # case single parameter
else: # case key = module
for (mname, pname), data in self.cache.items():
if mname == key:
cbfunc(mname, pname, *data)
elif cbname == 'nodeStateChange':
cbfunc(self.online, self._state)
if kwds:
raise TypeError('unknown callback: %s' % (', '.join(kwds)))
def callback(self, key, cbname, *args):
"""perform callbacks
key=None:
key=<module name>: callbacks for specified module
key=(<module name>, <parameter name): callbacks for specified parameter
"""
cblist = self.callbacks[cbname].get(key, [])
self.callbacks[cbname][key] = [cb for cb in cblist if cb(*args) is not UNREGISTER]
return bool(cblist)
def updateValue(self, module, param, value, timestamp, readerror):
if readerror:
assert isinstance(readerror, Exception)
if self.validate_data:
try:
# try to validate, reason: make enum_members from integers
datatype = self.modules[module]['parameters'][param]['datatype']
value = datatype(value)
except (KeyError, ValueError):
pass
self.cache[(module, param)] = (value, timestamp, readerror)
self.callback(None, 'updateEvent', module, param, value, timestamp, readerror)
self.callback(module, 'updateEvent', module, param, value, timestamp, readerror)
self.callback((module, param), 'updateEvent', module, param, value, timestamp, readerror)
def getParameter(self, module, parameter, trycache=False):
if trycache:
cached = self.cache.get((module, parameter), None)
if cached:
return cached
if self.online:
self.readParameter(module, parameter)
return self.cache[module, parameter]
def readParameter(self, module, parameter):
"""forced read over connection"""
raise NotImplementedError()
class SecopClient(ProxyClient):
"""a general SECoP client"""
reconnect_timeout = 10
shutdown = False
_rxthread = None
_txthread = None
disconnect_time = 0 # time of last disconnect
secop_version = ''
descriptive_data = {}
modules = {}
_last_error = None
def __init__(self, uri, log=Logger):
super().__init__()
# maps expected replies to [request, Event, is_error, result] until a response came
# there can only be one entry per thread calling 'request'
self.active_requests = {}
self.io = None
self.txq = queue.Queue(30) # queue for tx requests
self.pending = queue.Queue(30) # requests with colliding action + ident
self.log = log
self.uri = uri
self.nodename = uri
self._lock = RLock()
def __del__(self):
try:
self.disconnect()
except Exception:
pass
def connect(self, try_period=0):
"""establish connection
if a <try_period> is given, repeat trying for the given time (sec)
"""
with self._lock:
if self.io:
return
if self.online:
self._set_state(True, 'reconnecting')
else:
self._set_state(False, 'connecting')
deadline = time.time() + try_period
while True:
try:
self.io = AsynConn(self.uri) # timeout 1 sec
self.io.writeline(IDENTREQUEST.encode('utf-8'))
reply = self.io.readline(10)
if reply:
self.secop_version = reply.decode('utf-8')
else:
raise self.error_map('HardwareError')('no answer to %s' % IDENTREQUEST)
if not self.secop_version.startswith(IDENTPREFIX):
raise self.error_map('HardwareError')('bad answer to %s: %r' %
(IDENTREQUEST, self.secop_version))
# now its safe to do secop stuff
self._rxthread = mkthread(self.__rxthread)
self._txthread = mkthread(self.__txthread)
self.log.debug('connected to %s', self.uri)
# pylint: disable=unsubscriptable-object
self._init_descriptive_data(self.request(DESCRIPTIONREQUEST)[2])
self.nodename = self.properties.get('equipment_id', self.uri)
if self.activate:
self.request(ENABLEEVENTSREQUEST)
self._set_state(True, 'connected')
break
except Exception:
# print(formatExtendedTraceback())
if time.time() > deadline:
# stay online for now, if activated
self._set_state(self.online and self.activate, 'disconnected')
raise
time.sleep(1)
self.log.info('%s ready', self.nodename)
def __txthread(self):
while not self.shutdown:
entry = self.txq.get()
if entry is None:
break
request = entry[0]
reply_action = REQUEST2REPLY.get(request[0], None)
if reply_action:
key = (reply_action, request[1]) # action and identifier
else: # allow experimental unknown requests, but only one at a time
key = None
if key in self.active_requests:
# store to requeue after the next reply was received
self.pending.put(entry)
else:
self.active_requests[key] = entry
line = encode_msg_frame(*request)
self.log.debug('TX: %r', line)
self.io.send(line)
self._txthread = None
self.disconnect()
def __rxthread(self):
while not self.shutdown:
try:
reply = self.io.readline()
if reply is None:
continue
except ConnectionClosed:
break
action, ident, data = decode_msg(reply)
if ident == '.':
ident = None
if action in UPDATE_MESSAGES:
module_param = self.internal.get(ident, None)
if module_param is None and ':' not in ident:
# allow missing ':value'/':target'
if action == WRITEREPLY:
module_param = self.internal.get(ident + ':target', None)
else:
module_param = self.internal.get(ident + ':value', None)
if module_param is not None:
if action.startswith(ERRORPREFIX):
timestamp = data[2].get('t', None)
readerror = secop.errors.make_secop_error(*data[0:2])
value = None
else:
timestamp = data[1].get('t', None)
value = data[0]
readerror = None
module, param = module_param
self.updateValue(module, param, value, timestamp, readerror)
if action in (EVENTREPLY, ERRORPREFIX + EVENTREPLY):
continue
try:
key = action, ident
entry = self.active_requests.pop(key)
except KeyError:
if action.startswith(ERRORPREFIX):
try:
key = REQUEST2REPLY[action[len(ERRORPREFIX):]], ident
except KeyError:
key = None
entry = self.active_requests.pop(key, None)
else:
# this may be a response to the last unknown request
key = None
entry = self.active_requests.pop(key, None)
if entry is None:
self._unhandled_message(action, ident, data)
continue
entry[2] = action, ident, data
entry[1].set() # trigger event
while not self.pending.empty():
# let the TX thread sort out which entry to treat
# this may have bad performance, but happens rarely
self.txq.put(self.pending.get())
self._rxthread = None
self.disconnect()
if self.activate:
self.log.info('reconnect to %s', self.uri)
mkthread(self._reconnect)
else:
self.log.warning('%s disconnected', self.uri)
self._set_state(False, 'disconnected')
def spawn_connect(self, connected_callback=None):
"""try to connect in background
and trigger event when done and event is not None
"""
self.disconnect_time = time.time()
mkthread(self._reconnect, connected_callback)
def _reconnect(self, connected_callback=None):
while True:
try:
self.connect()
if connected_callback:
connected_callback()
break
except Exception as e:
txt = str(e).split('\n', 1)[0]
if txt != self._last_error:
self._last_error = txt
self.log.error(str(e))
if time.time() > self.disconnect_time + self.reconnect_timeout:
if self.online: # was recently connected
self.disconnect_time = 0
self.log.warning('can not reconnect to %s (%r)' % (self.nodename, e))
# self.log.warning(formatExtendedTraceback())
self._set_state(False, 'disconnected')
time.sleep(self.reconnect_timeout)
else:
time.sleep(1)
def disconnect(self):
self.shutdown = True
self.disconnect_time = time.time()
if self._txthread:
self.txq.put(None) # shutdownmarker
self._txthread.join()
self._txthread = None
if self._rxthread:
self._rxthread.join()
self._rxthread = None
if self.io:
self.io.disconnect()
self.io = None
# abort pending requests early
try: # avoid race condition
while self.active_requests:
_, (_, event, _) = self.active_requests.popitem()
event.set()
except KeyError:
pass
try:
while True:
_, event, _ = self.pending.get(block=False)
event.set()
except queue.Empty:
pass
self.shutdown = False
def _init_descriptive_data(self, data):
"""rebuild descriptive data"""
changed_modules = None
if json.dumps(data, sort_keys=True) != json.dumps(self.descriptive_data, sort_keys=True):
if self.descriptive_data:
changed_modules = set()
modules = data.get('modules', {})
for modname, moddesc in self.descriptive_data['modules'].items():
if json.dumps(moddesc, sort_keys=True) != json.dumps(modules.get(modname), sort_keys=True):
changed_modules.add(modname)
self.descriptive_data = data
modules = data['modules']
self.modules = {}
self.properties = {k: v for k, v in data.items() if k != 'modules'}
self.identifier = {} # map (module, parameter) -> identifier
self.internal = {} # map identifier -> (module, parameter)
for modname, moddescr in modules.items():
# separate accessibles into command and parameters
parameters = {}
commands = {}
accessibles = moddescr['accessibles']
for aname, aentry in accessibles.items():
aentry = dict(aentry, datatype=get_datatype(aentry['datainfo']))
iname = self.internalize_name(aname)
ident = '%s:%s' % (modname, aname)
self.identifier[modname, iname] = ident
self.internal[ident] = modname, iname
if aentry['datainfo']['type'] == 'command':
commands[iname] = aentry
else:
parameters[iname] = aentry
properties = {k: v for k, v in moddescr.items() if k != 'accessibles'}
self.modules[modname] = dict(accessibles=accessibles, parameters=parameters,
commands=commands, properties=properties)
if changed_modules is not None:
done = self.callback(None, 'descriptiveDataChange', None, self)
for mname in changed_modules:
if not self.callback(mname, 'descriptiveDataChange', mname, self):
self.log.warning('descriptive data changed on module %r', mname)
done = True
if not done:
self.log.warning('descriptive data of %r changed', self.nodename)
def _unhandled_message(self, action, ident, data):
if not self.callback(None, 'unhandledMessage', action, ident, data):
self.log.warning('unhandled message: %s %s %r' % (action, ident, data))
def _set_state(self, online, state=None):
# treat reconnecting as online!
self._state = state or self._state
self.online = online
self.callback(None, 'nodeStateChange', self.online, self._state)
for mname in self.modules:
self.callback(mname, 'nodeStateChange', self.online, self._state)
def queue_request(self, action, ident=None, data=None):
"""make a request"""
request = action, ident, data
self.connect() # make sure we are connected
# the last item is for the reply
entry = [request, Event(), None]
self.txq.put(entry)
return entry
def get_reply(self, entry):
"""wait for reply and return it"""
if not entry[1].wait(10): # event
raise TimeoutError('no response within 10s')
if not entry[2]: # reply
raise ConnectionError('connection closed before reply')
action, _, data = entry[2] # pylint: disable=unpacking-non-sequence
if action.startswith(ERRORPREFIX):
errcls = self.error_map(data[0])
raise errcls(data[1])
return entry[2] # reply
def request(self, action, ident=None, data=None):
"""make a request
and wait for reply
"""
entry = self.queue_request(action, ident, data)
return self.get_reply(entry)
def readParameter(self, module, parameter):
try:
self.request(READREQUEST, self.identifier[module, parameter])
except secop.errors.SECoPError:
# error reply message is already stored as readerror in cache
pass
return self.cache.get((module, parameter), None)
def setParameter(self, module, parameter, value):
self.connect() # make sure we are connected
datatype = self.modules[module]['parameters'][parameter]['datatype']
value = datatype.export_value(datatype.from_string(value))
self.request(WRITEREQUEST, self.identifier[module, parameter], value)
return self.cache[module, parameter]
def execCommand(self, module, command, argument=None):
self.connect() # make sure we are connected
datatype = self.modules[module]['commands'][command]['datatype'].argument
if datatype:
argument = datatype.export_value(datatype.from_string(argument))
else:
if argument is not None:
raise secop.errors.BadValueError('command has no argument')
# pylint: disable=unsubscriptable-object
data, qualifiers = self.request(COMMANDREQUEST, self.identifier[module, command], argument)[2]
datatype = self.modules[module]['commands'][command]['datatype'].result
if datatype:
data = datatype.import_value(data)
return data, qualifiers
# the following attributes may be/are intended to be overwritten by a subclass
ERROR_MAP = secop.errors.EXCEPTIONS
DEFAULT_EXCEPTION = secop.errors.SECoPError
PREDEFINED_NAMES = set(secop.params.PREDEFINED_ACCESSIBLES)
activate = True
def error_map(self, exc):
"""how to convert SECoP and unknown exceptions"""
return self.ERROR_MAP.get(exc, self.DEFAULT_EXCEPTION)
def internalize_name(self, name):
"""how to create internal names"""
if name.startswith('_') and name[1:] not in self.PREDEFINED_NAMES:
return name[1:]
return name
Reference in New Issue View Git Blame Copy Permalink