Source code for nxswriter.ClientSource

#!/usr/bin/env python
#   This file is part of nexdatas - Tango Server for NeXus data writer
#
#    Copyright (C) 2012-2017 DESY, Jan Kotanski <jkotan@mail.desy.de>
#
#    nexdatas 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 3 of the License, or
#    (at your option) any later version.
#
#    nexdatas 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 nexdatas.  If not, see <http://www.gnu.org/licenses/>.
#

""" Definitions of CLIENT datasource """

import sys
import xml.etree.ElementTree as et
from lxml.etree import XMLParser
from .DataSources import DataSource
from .Errors import DataSourceSetupError


[docs]class ClientSource(DataSource): """ Client data source """ def __init__(self, streams=None, name=None): """ constructor :brief: It sets all member variables to None :param streams: tango-like steamset class :type streams: :class:`StreamSet` or :class:`tango.LatestDeviceImpl` :param name: datasource name :type name: :obj:`str` """ DataSource.__init__(self, streams=streams, name=name) #: (:obj:`str`) data name self.name = None #: (:obj:`dict` <:obj:`str`, :obj:`dict` <:obj:`str`, any>>) #: the current static JSON object self.__globalJSON = None #: (:obj:`dict` <:obj:`str`, :obj:`dict` <:obj:`str`, any>>) #: the current dynamic JSON object self.__localJSON = None
[docs] def setup(self, xml): """ sets the parrameters up from xml :param xml: datasource parameters :type xml: :obj:`str` :raises: :exc:`nxswriter.Errors.DataSourceSetupError` \ : if :obj:`name` is not defined """ if sys.version_info > (3,): xml = bytes(xml, "UTF-8") root = et.fromstring(xml, parser=XMLParser(collect_ids=False)) rec = root.find("record") if rec is not None: self.name = rec.get("name") if not self.name: if self._streams: self._streams.error( "ClientSource::setup() - " "Client record name not defined: %s" % xml, std=False) raise DataSourceSetupError( "Client record name not defined: %s" % xml)
def __str__(self): """ self-description :returns: self-describing string :rtype: :obj:`str` """ return " CLIENT record %s" % (self.name)
[docs] def setJSON(self, globalJSON, localJSON=None): """ sets JSON string :brief: It sets the currently used JSON string :param globalJSON: static JSON string :type globalJSON: :obj:`dict` \ : <:obj:`str`, :obj:`dict` <:obj:`str`, any>> :param localJSON: dynamic JSON string :type localJSON: :obj:`dict` \ <:obj:`str`, :obj:`dict` <:obj:`str`, any>> """ self.__globalJSON = globalJSON self.__localJSON = localJSON
[docs] def getData(self): """ provides access to the data :returns: dictionary with collected data :rtype: {'rank': :obj:`str`, 'value': any, 'tangoDType': :obj:`str`, \ : 'shape': :obj:`list` <int>, 'encoding': :obj:`str`, \ : 'decoders': :obj:`str`} ) """ names = [self.name] if self.name: names.append(self.name.lower()) return self._getJSONData(names, self.__globalJSON, self.__localJSON)