summaryrefslogtreecommitdiff
path: root/silx/resources/__init__.py
diff options
context:
space:
mode:
authorAlexandre Marie <alexandre.marie@synchrotron-soleil.fr>2019-07-09 10:20:20 +0200
committerAlexandre Marie <alexandre.marie@synchrotron-soleil.fr>2019-07-09 10:20:20 +0200
commit654a6ac93513c3cc1ef97cacd782ff674c6f4559 (patch)
tree3b986e4972de7c57fa465820367602fc34bcb0d3 /silx/resources/__init__.py
parenta763e5d1b3921b3194f3d4e94ab9de3fbe08bbdd (diff)
New upstream version 0.11.0+dfsg
Diffstat (limited to 'silx/resources/__init__.py')
-rw-r--r--silx/resources/__init__.py200
1 files changed, 3 insertions, 197 deletions
diff --git a/silx/resources/__init__.py b/silx/resources/__init__.py
index ba390d6..5346f48 100644
--- a/silx/resources/__init__.py
+++ b/silx/resources/__init__.py
@@ -56,20 +56,14 @@ of this modules to ensure access across different distribution schemes:
__authors__ = ["V.A. Sole", "Thomas Vincent", "J. Kieffer"]
__license__ = "MIT"
-__date__ = "13/02/2019"
+__date__ = "08/03/2019"
import os
import sys
-import threading
-import json
import logging
-import tempfile
-import unittest
import importlib
-import six
-
logger = logging.getLogger(__name__)
@@ -288,193 +282,5 @@ def _resource_filename(resource, default_directory=None):
return pkg_resources.resource_filename(package_name, resource_name)
-class ExternalResources(object):
- """Utility class which allows to download test-data from www.silx.org
- and manage the temporary data during the tests.
-
- """
-
- def __init__(self, project,
- url_base,
- env_key=None,
- timeout=60):
- """Constructor of the class
-
- :param str project: name of the project, like "silx"
- :param str url_base: base URL for the data, like "http://www.silx.org/pub"
- :param str env_key: name of the environment variable which contains the
- test_data directory, like "SILX_DATA".
- If None (default), then the name of the
- environment variable is built from the project argument:
- "<PROJECT>_DATA".
- The environment variable is optional: in case it is not set,
- a directory in the temporary folder is used.
- :param timeout: time in seconds before it breaks
- """
- self.project = project
- self._initialized = False
- self.sem = threading.Semaphore()
-
- self.env_key = env_key or (self.project.upper() + "_TESTDATA")
- self.url_base = url_base
- self.all_data = set()
- self.timeout = timeout
- self._data_home = None
-
- @property
- def data_home(self):
- """Returns the data_home path and make sure it exists in the file
- system."""
- if self._data_home is not None:
- return self._data_home
-
- data_home = os.environ.get(self.env_key)
- if data_home is None:
- try:
- import getpass
- name = getpass.getuser()
- except Exception:
- if "getlogin" in dir(os):
- name = os.getlogin()
- elif "USER" in os.environ:
- name = os.environ["USER"]
- elif "USERNAME" in os.environ:
- name = os.environ["USERNAME"]
- else:
- name = "uid" + str(os.getuid())
-
- basename = "%s_testdata_%s" % (self.project, name)
- data_home = os.path.join(tempfile.gettempdir(), basename)
- if not os.path.exists(data_home):
- os.makedirs(data_home)
- self._data_home = data_home
- return data_home
-
- def _initialize_data(self):
- """Initialize for downloading test data"""
- if not self._initialized:
- with self.sem:
- if not self._initialized:
- self.testdata = os.path.join(self.data_home, "all_testdata.json")
- if os.path.exists(self.testdata):
- with open(self.testdata) as f:
- self.all_data = set(json.load(f))
- self._initialized = True
-
- def clean_up(self):
- pass
-
- def getfile(self, filename):
- """Downloads the requested file from web-server available
- at https://www.silx.org/pub/silx/
-
- :param: relative name of the image.
- :return: full path of the locally saved file.
- """
- logger.debug("ExternalResources.getfile('%s')", filename)
-
- if not self._initialized:
- self._initialize_data()
-
- fullfilename = os.path.abspath(os.path.join(self.data_home, filename))
-
- if not os.path.isfile(fullfilename):
- logger.debug("Trying to download image %s, timeout set to %ss",
- filename, self.timeout)
- dictProxies = {}
- if "http_proxy" in os.environ:
- dictProxies['http'] = os.environ["http_proxy"]
- dictProxies['https'] = os.environ["http_proxy"]
- if "https_proxy" in os.environ:
- dictProxies['https'] = os.environ["https_proxy"]
- if dictProxies:
- proxy_handler = six.moves.urllib.request.ProxyHandler(dictProxies)
- opener = six.moves.urllib.request.build_opener(proxy_handler).open
- else:
- opener = six.moves.urllib.request.urlopen
-
- logger.debug("wget %s/%s", self.url_base, filename)
- try:
- data = opener("%s/%s" % (self.url_base, filename),
- data=None, timeout=self.timeout).read()
- logger.info("Image %s successfully downloaded.", filename)
- except six.moves.urllib.error.URLError:
- raise unittest.SkipTest("network unreachable.")
-
- if not os.path.isdir(os.path.dirname(fullfilename)):
- # Create sub-directory if needed
- os.makedirs(os.path.dirname(fullfilename))
-
- try:
- with open(fullfilename, "wb") as outfile:
- outfile.write(data)
- except IOError:
- raise IOError("unable to write downloaded \
- data to disk at %s" % self.data_home)
-
- if not os.path.isfile(fullfilename):
- raise RuntimeError(
- "Could not automatically \
- download test images %s!\n \ If you are behind a firewall, \
- please set both environment variable http_proxy and https_proxy.\
- This even works under windows ! \n \
- Otherwise please try to download the images manually from \n%s/%s"
- % (filename, self.url_base, filename))
-
- if filename not in self.all_data:
- self.all_data.add(filename)
- image_list = list(self.all_data)
- image_list.sort()
- try:
- with open(self.testdata, "w") as fp:
- json.dump(image_list, fp, indent=4)
- except IOError:
- logger.debug("Unable to save JSON list")
-
- return fullfilename
-
- def getdir(self, dirname):
- """Downloads the requested tarball from the server
- https://www.silx.org/pub/silx/
- and unzips it into the data directory
-
- :param: relative name of the image.
- :return: list of files with their full path.
- """
- lodn = dirname.lower()
- if (lodn.endswith("tar") or lodn.endswith("tgz") or
- lodn.endswith("tbz2") or lodn.endswith("tar.gz") or
- lodn.endswith("tar.bz2")):
- import tarfile
- engine = tarfile.TarFile.open
- elif lodn.endswith("zip"):
- import zipfile
- engine = zipfile.ZipFile
- else:
- raise RuntimeError("Unsupported archive format. Only tar and zip "
- "are currently supported")
- full_path = self.getfile(dirname)
- root = os.path.dirname(full_path)
- with engine(full_path, mode="r") as fd:
- fd.extractall(self.data_home)
- if lodn.endswith("zip"):
- result = [os.path.join(root, i) for i in fd.namelist()]
- else:
- result = [os.path.join(root, i) for i in fd.getnames()]
- return result
-
- def download_all(self, imgs=None):
- """Download all data needed for the test/benchmarks
-
- :param imgs: list of files to download, by default all
- :return: list of path with all files
- """
- if not self._initialized:
- self._initialize_data()
- if not imgs:
- imgs = self.all_data
- res = []
- for fn in imgs:
- logger.info("Downloading from silx.org: %s", fn)
- res.append(self.getfile(fn))
- return res
+# Expose ExternalResources for compatibility (since silx 0.11)
+from ..utils.ExternalResources import ExternalResources