#!/usr/bin/env python
# ###########################################################################
#
# This file is part of Taurus
#
# http://taurus-scada.org
#
# Copyright 2011 CELLS / ALBA Synchrotron, Bellaterra, Spain
#
# Taurus is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Taurus 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Taurus. If not, see <http://www.gnu.org/licenses/>.
#
# ###########################################################################
"""
resfactory.py:
"""
import os
import importlib
from collections.abc import Mapping
from taurus import Manager
from taurus.core.util.singleton import Singleton
from taurus.core.util.log import Logger
from taurus.core.taurusfactory import TaurusFactory
from taurus.core.taurusexception import TaurusException
[docs]
class ResourcesFactory(Singleton, TaurusFactory, Logger):
"""A Singleton class designed to provide Simulation related objects."""
#: the list of schemes that this factory supports. For this factory: 'res'
#: and 'resources' are the supported schemes
schemes = (
"res",
"resource",
)
#: the default resource file name
DftResourceName = "taurus_resources.py"
#: priority for the default resource
DftResourcePriority = 10
def __init__(self):
"""Initialization. Nothing to be done here for now."""
pass
[docs]
def init(self, *args, **kwargs):
"""Singleton instance initialization.
**For internal usage only**
"""
name = self.__class__.__name__
self.call__init__(Logger, name)
self.call__init__(TaurusFactory)
self.clear()
[docs]
def clear(self):
self._resource_map = {}
self._resource_priority = {}
self._resource_priority_keys = []
self._resource_count = 0
[docs]
def reloadResource(self, obj=None, priority=1, name=None):
"""(Re)Loads the given resource.
:param obj: the resource object. Default is None meaning in will
(re)load the default resource: taurus_resources.py from the
application directory
:type obj: dict or file or None
:param priority: the resource priority. Default is 1 meaning maximum
priority
:type priority: int
:param name: an optional name to give to the resource
:type name: str
:return: a dictionary version of the given resource object
:rtype: dict
"""
if priority < 1:
raise ValueError("priority must be >=1")
if isinstance(obj, Mapping):
name = name or "DICT%02d" % priority
elif type(obj) in (str,) or obj is None:
name, mod = self.__reloadResource(obj)
obj = {}
for k, v in mod.__dict__.items():
if not k.startswith("_") and isinstance(v, str):
obj[k] = v
else:
raise TypeError
if self._resource_map.get(name) is None:
self._resource_count += 1
self._resource_map[name] = obj
if self._resource_count == 1:
self._first_resource = obj
pl = self._resource_priority.get(priority)
if pl is None:
self._resource_priority[priority] = pl = []
pl.append(name)
self._resource_priority_keys = list(self._resource_priority.keys())
self._resource_priority_keys.sort()
return obj
loadResource = reloadResource
loadResource.__doc__ = reloadResource.__doc__
def __reloadResource(self, name=None):
path = os.path.curdir
if name is None:
file_name = ResourcesFactory.DftResourceName
else:
path, file_name = os.path.split(name)
if not path:
path = os.path.curdir
path = os.path.abspath(path)
full_name = os.path.join(path, file_name)
if not os.path.isfile(full_name):
raise ImportError
module_name, ext = os.path.splitext(file_name)
# import module from path
_spec = importlib.util.spec_from_file_location(module_name, full_name)
m = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(m)
return full_name, m
[docs]
def getValue(self, key):
"""Returns the value for a given key
:param key: a key
:type key: str
:return: the value for the given key
:rtype: str
"""
if self._resource_count == 0:
try:
self.reloadResource(priority=self.DftResourcePriority)
except Exception:
return None
# optimization: many applications contain only one resource: in that
# case avoid the loop
if self._resource_count == 1:
return self._first_resource.get(key, None)
for p in self._resource_priority_keys:
for resource_name in self._resource_priority[p]:
resource = self._resource_map[resource_name]
try:
return resource[key]
except Exception:
pass
[docs]
def findObjectClass(self, absolute_name):
"""
Obtain the class object corresponding to the given name.
:param absolute_name: the object absolute name string
:type absolute_name: str
:return: the class for the model object mapped by absolute_name, or
None if absolute_name is invalid.
:rtype: taurus.core.taurusmodel.TaurusModel or None
"""
validators = (
self.getAttributeNameValidator(),
self.getDeviceNameValidator(),
self.getAuthorityNameValidator(),
)
for v in validators:
try:
value = self.getValue(
v.getUriGroups(absolute_name)["_resname"]
)
return Manager().findObjectClass(value)
except Exception:
pass
return None
[docs]
def getAuthority(self, name=None):
"""
Obtain the authority model object referenced by name.
:param name: name
:type name: str
:return: authority object :raise:
(taurus.core.taurusexception.TaurusException) if name is invalid
:rtype: taurus.core.taurusauthority.TaurusAuthority
"""
groups = self.getAuthorityNameValidator().getUriGroups(name)
if groups is None:
raise TaurusException('Invalid name "%s"' % name)
res_name = groups["_resname"]
value = self.getValue(res_name)
return Manager().getAuthority(value)
[docs]
def getDevice(self, name):
"""
Obtain the device model object referenced by name.
:param name: name
:type name: str
:return: device object :raise:
(taurus.core.taurusexception.TaurusException) if name is invalid
:rtype: taurus.core.taurusdevice.TaurusDevice
"""
groups = self.getDeviceNameValidator().getUriGroups(name)
if groups is None:
raise TaurusException('Invalid name "%s"' % name)
res_name = groups["_resname"]
value = self.getValue(res_name)
return Manager().getDevice(value)
[docs]
def getAttribute(self, name):
"""
Obtain the attribute model object referenced by name.
:param name: name
:type name: str
:return: attribute object :raise:
(taurus.core.taurusexception.TaurusException) if name is invalid
:rtype: taurus.core.taurusattribute.TaurusAttribute
"""
groups = self.getAttributeNameValidator().getUriGroups(name)
if groups is None:
raise TaurusException('Invalid name "%s"' % name)
res_name = groups["_resname"]
value = self.getValue(res_name)
return Manager().getAttribute(value)
[docs]
def getAuthorityNameValidator(self):
"""Return ResourceAuthorityNameValidator"""
from . import resvalidator
return resvalidator.ResourceAuthorityNameValidator()
[docs]
def getDeviceNameValidator(self):
"""Return ResourceDeviceNameValidator"""
from . import resvalidator
return resvalidator.ResourceDeviceNameValidator()
[docs]
def getAttributeNameValidator(self):
"""Return ResourceAttributeNameValidator"""
from . import resvalidator
return resvalidator.ResourceAttributeNameValidator()