11857958346

Committed 15 Nov 2024 02:23PM UTC coverage: 88.388% (-0.01%) from 88.401%

Build # 11857958346

Build Type

Pull #649

github

Committed by

web-flow

Commit Message

Merge b21bfe1e7 into 123ca040b

Pull Request Pull Request #649: [Fix] instantaneous rate with trials

Run Details

6729 of 7613 relevant lines covered (88.39%)

4.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.68

/elephant/datasets.py

import hashlib
import os
import ssl
import tempfile
from urllib.parse import urlparse
import warnings
from os import environ, getenv
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import urlopen, urlretrieve
from zipfile import ZipFile

from tqdm import tqdm

from elephant import _get_version

ELEPHANT_TMP_DIR = Path(tempfile.gettempdir()) / "elephant"


class TqdmUpTo(tqdm):
    """
    Provides `update_to(n)` which uses `tqdm.update(delta_n)`.
    Original implementation:
    https://github.com/tqdm/tqdm/blob/master/examples/tqdm_wget.py
    """

    def update_to(self, b=1, bsize=1, tsize=None):
        """
        b : int, optional
            Number of blocks transferred so far [default: 1].
        bsize : int, optional
            Size of each block (in tqdm units) [default: 1].
        tsize : int, optional
            Total size (in tqdm units). If [default: None] remains unchanged.
        """
        if tsize is not None:
            self.total = tsize
        self.update(b * bsize - self.n)  # will also set self.n = b * bsize


def calculate_md5(filepath, chunk_size=1024 * 1024):
    md5 = hashlib.md5()
    with open(filepath, 'rb') as f:
        for chunk in iter(lambda: f.read(chunk_size), b''):
            md5.update(chunk)
    return md5.hexdigest()


def check_integrity(filepath, md5):
    if not Path(filepath).exists() or md5 is None:
        return False
    return calculate_md5(filepath) == md5


def download(url, filepath=None, checksum=None, verbose=True):
    if filepath is None:
        filename = url.split('/')[-1]
        filepath = ELEPHANT_TMP_DIR / filename
    filepath = Path(filepath)
    if check_integrity(filepath, md5=checksum):
        return filepath
    folder = filepath.absolute().parent
    folder.mkdir(exist_ok=True)
    desc = f"Downloading {url} to '{filepath}'"
    with TqdmUpTo(unit='B', unit_scale=True, unit_divisor=1024, miniters=1,
                  desc=desc, disable=not verbose) as t:
        try:
            urlretrieve(url, filename=filepath, reporthook=t.update_to)
        except URLError:
            # do not authenticate SSL certificate
            ssl._create_default_https_context = ssl._create_unverified_context
            urlretrieve(url, filename=filepath, reporthook=t.update_to)
    return filepath


def download_datasets(repo_path, filepath=None, checksum=None,
                      verbose=True):
    r"""
    This function can be used to download files from elephant-data using
    only the path relative to the root of the elephant-data repository.
    The default URL used, points to elephants corresponding release of
    elephant-data.
    Different versions of the elephant package may require different
    versions of elephant-data.
    e.g. the following URLs:
    -  https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/0.0.1
       points to release v0.0.1.
    -  https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/master
       always points to the latest state of elephant-data.
    -  https://datasets.python-elephant.org/
       points to the root of elephant data

    To change this URL, use the environment variable `ELEPHANT_DATA_LOCATION`.
    When using data, which is not yet contained in the master branch or a
    release of elephant data, e.g. during development, this variable can
    be used to change the default URL.
    For example to use data on branch `multitaper`, change the
    `ELEPHANT_DATA_LOCATION` to
    https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/multitaper.
    For a complete example, see Examples section.
        
    To use a local copy of elephant-data, use the environment variable
    `ELEPHANT_DATA_LOCATION`, e.g. set to /home/user/elephant-data.
        
    Parameters
    ----------
    repo_path : str
        String denoting the path relative to elephant-data repository root
    filepath : str, optional
        Path to temporary folder where the downloaded files will be stored
    checksum : str, optional
        Checksum to verify data integrity after download
    verbose : bool, optional
        Whether to disable the entire progressbar wrapper [].
        If set to None, disable on non-TTY.
        Default: True

    Returns
    -------
    filepath : pathlib.Path
        Path to downloaded files.


    Notes
    -----
    The default URL always points to elephant-data. Please
    do not change its value. For development purposes use the environment
    variable 'ELEPHANT_DATA_LOCATION'.

    Examples
    --------
    The following example downloads a file from elephant-data branch
    'multitaper', by setting the environment variable to the branch URL:

    >>> import os
    >>> from elephant.datasets import download_datasets
    >>> os.environ["ELEPHANT_DATA_LOCATION"] = "https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/multitaper" # noqa
    >>> download_datasets("unittest/spectral/multitaper_psd/data/time_series.npy") # doctest: +SKIP
    PosixPath('/tmp/elephant/time_series.npy')
    """

    env_var = 'ELEPHANT_DATA_LOCATION'
    if env_var in os.environ:  # user did set path or URL
        if os.path.exists(getenv(env_var)):
            return Path(f"{getenv(env_var)}/{repo_path}")
        elif urlparse(getenv(env_var)).scheme not in ('http', 'https'):
            raise ValueError(f"The environment variable {env_var} must be set to either an existing file system path "
                             f"or a valid URL. Given value: '{getenv(env_var)}' is neither.")

    # this url redirects to the current location of elephant-data
    url_to_root = "https://datasets.python-elephant.org/"

    # get URL to corresponding version of elephant data
    # (version elephant is equal to version elephant-data)
    default_url = url_to_root + f"raw/v{_get_version()}"

    if env_var not in environ:  # user did not set URL
        # is 'version-URL' available? (not for elephant development version)
        try:
            urlopen(default_url+'/README.md')

        except HTTPError as error:
            # if corresponding elephant-data version is not found,
            # use latest commit of elephant-data
            default_url = url_to_root + "raw/master"

            warnings.warn(f"No corresponding version of elephant-data found.\n"
                          f"Elephant version: {_get_version()}. "
                          f"Data URL:{error.url}, error: {error}.\n"
                          f"Using elephant-data latest instead (This is "
                          f"expected for elephant development versions).")

        except URLError as error:
            # if verification of SSL certificate fails, do not verify cert
            try:  # try again without certificate verification
                ctx = ssl._create_unverified_context()
                ctx.check_hostname = True
                urlopen(default_url + '/README.md')
            except HTTPError:  # e.g. 404
                default_url = url_to_root + "raw/master"

            warnings.warn(f"Data URL:{default_url}, error: {error}."
                          f"{error.reason}")

    url = f"{getenv(env_var, default_url)}/{repo_path}"

    return download(url, filepath, checksum, verbose)


def unzip(filepath, outdir=ELEPHANT_TMP_DIR, verbose=True):
    with ZipFile(filepath) as zfile:
        zfile.extractall(path=outdir)
    if verbose:
        print(f"Extracted {filepath} to {outdir}")

1	import hashlib	5✔
2	import os	5✔
3	import ssl	5✔
4	import tempfile	5✔
5	from urllib.parse import urlparse	5✔
6	import warnings	5✔
7	from os import environ, getenv	5✔
8	from pathlib import Path	5✔
9	from urllib.error import HTTPError, URLError	5✔
10	from urllib.request import urlopen, urlretrieve	5✔
11	from zipfile import ZipFile	5✔
12
13	from tqdm import tqdm	5✔
14
15	from elephant import _get_version	5✔
16
17	ELEPHANT_TMP_DIR = Path(tempfile.gettempdir()) / "elephant"	5✔
18
19
20	class TqdmUpTo(tqdm):	5✔
21	"""
22	Provides `update_to(n)` which uses `tqdm.update(delta_n)`.
23	Original implementation:
24	https://github.com/tqdm/tqdm/blob/master/examples/tqdm_wget.py
25	"""
26
27	def update_to(self, b=1, bsize=1, tsize=None):	5✔
28	"""
29	b : int, optional
30	Number of blocks transferred so far [default: 1].
31	bsize : int, optional
32	Size of each block (in tqdm units) [default: 1].
33	tsize : int, optional
34	Total size (in tqdm units). If [default: None] remains unchanged.
35	"""
36	if tsize is not None:	5✔
37	self.total = tsize	5✔
38	self.update(b * bsize - self.n) # will also set self.n = b * bsize	5✔
39
40
41	def calculate_md5(filepath, chunk_size=1024 * 1024):	5✔
42	md5 = hashlib.md5()	×
43	with open(filepath, 'rb') as f:	×
44	for chunk in iter(lambda: f.read(chunk_size), b''):	×
45	md5.update(chunk)	×
46	return md5.hexdigest()	×
47
48
49	def check_integrity(filepath, md5):	5✔
50	if not Path(filepath).exists() or md5 is None:	5✔
51	return False	5✔
52	return calculate_md5(filepath) == md5	×
53
54
55	def download(url, filepath=None, checksum=None, verbose=True):	5✔
56	if filepath is None:	5✔
57	filename = url.split('/')[-1]	5✔
58	filepath = ELEPHANT_TMP_DIR / filename	5✔
59	filepath = Path(filepath)	5✔
60	if check_integrity(filepath, md5=checksum):	5✔
61	return filepath	×
62	folder = filepath.absolute().parent	5✔
63	folder.mkdir(exist_ok=True)	5✔
64	desc = f"Downloading {url} to '{filepath}'"	5✔
65	with TqdmUpTo(unit='B', unit_scale=True, unit_divisor=1024, miniters=1,	5✔
66	desc=desc, disable=not verbose) as t:
67	try:	5✔
68	urlretrieve(url, filename=filepath, reporthook=t.update_to)	5✔
69	except URLError:	5✔
70	# do not authenticate SSL certificate
71	ssl._create_default_https_context = ssl._create_unverified_context	5✔
72	urlretrieve(url, filename=filepath, reporthook=t.update_to)	5✔
73	return filepath	5✔
74
75
76	def download_datasets(repo_path, filepath=None, checksum=None,	5✔
77	verbose=True):
78	r"""
79	This function can be used to download files from elephant-data using
80	only the path relative to the root of the elephant-data repository.
81	The default URL used, points to elephants corresponding release of
82	elephant-data.
83	Different versions of the elephant package may require different
84	versions of elephant-data.
85	e.g. the following URLs:
86	- https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/0.0.1
87	points to release v0.0.1.
88	- https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/master
89	always points to the latest state of elephant-data.
90	- https://datasets.python-elephant.org/
91	points to the root of elephant data
92
93	To change this URL, use the environment variable `ELEPHANT_DATA_LOCATION`.
94	When using data, which is not yet contained in the master branch or a
95	release of elephant data, e.g. during development, this variable can
96	be used to change the default URL.
97	For example to use data on branch `multitaper`, change the
98	`ELEPHANT_DATA_LOCATION` to
99	https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/multitaper.
100	For a complete example, see Examples section.
101
102	To use a local copy of elephant-data, use the environment variable
103	`ELEPHANT_DATA_LOCATION`, e.g. set to /home/user/elephant-data.
104
105	Parameters
106	----------
107	repo_path : str
108	String denoting the path relative to elephant-data repository root
109	filepath : str, optional
110	Path to temporary folder where the downloaded files will be stored
111	checksum : str, optional
112	Checksum to verify data integrity after download
113	verbose : bool, optional
114	Whether to disable the entire progressbar wrapper [].
115	If set to None, disable on non-TTY.
116	Default: True
117
118	Returns
119	-------
120	filepath : pathlib.Path
121	Path to downloaded files.
122
123
124	Notes
125	-----
126	The default URL always points to elephant-data. Please
127	do not change its value. For development purposes use the environment
128	variable 'ELEPHANT_DATA_LOCATION'.
129
130	Examples
131	--------
132	The following example downloads a file from elephant-data branch
133	'multitaper', by setting the environment variable to the branch URL:
134
135	>>> import os
136	>>> from elephant.datasets import download_datasets
137	>>> os.environ["ELEPHANT_DATA_LOCATION"] = "https://web.gin.g-node.org/NeuralEnsemble/elephant-data/raw/multitaper" # noqa
138	>>> download_datasets("unittest/spectral/multitaper_psd/data/time_series.npy") # doctest: +SKIP
139	PosixPath('/tmp/elephant/time_series.npy')
140	"""
141
142	env_var = 'ELEPHANT_DATA_LOCATION'	5✔
143	if env_var in os.environ: # user did set path or URL	5✔
144	if os.path.exists(getenv(env_var)):	5✔
145	return Path(f"{getenv(env_var)}/{repo_path}")	5✔
146	elif urlparse(getenv(env_var)).scheme not in ('http', 'https'):	5✔
147	raise ValueError(f"The environment variable {env_var} must be set to either an existing file system path "	5✔
148	f"or a valid URL. Given value: '{getenv(env_var)}' is neither.")
149
150	# this url redirects to the current location of elephant-data
151	url_to_root = "https://datasets.python-elephant.org/"	5✔
152
153	# get URL to corresponding version of elephant data
154	# (version elephant is equal to version elephant-data)
155	default_url = url_to_root + f"raw/v{_get_version()}"	5✔
156
157	if env_var not in environ: # user did not set URL	5✔
158	# is 'version-URL' available? (not for elephant development version)
159	try:	×
160	urlopen(default_url+'/README.md')	×
161
162	except HTTPError as error:	×
163	# if corresponding elephant-data version is not found,
164	# use latest commit of elephant-data
165	default_url = url_to_root + "raw/master"	×
166
167	warnings.warn(f"No corresponding version of elephant-data found.\n"	×
168	f"Elephant version: {_get_version()}. "
169	f"Data URL:{error.url}, error: {error}.\n"
170	f"Using elephant-data latest instead (This is "
171	f"expected for elephant development versions).")
172
173	except URLError as error:	×
174	# if verification of SSL certificate fails, do not verify cert
175	try: # try again without certificate verification	×
176	ctx = ssl._create_unverified_context()	×
177	ctx.check_hostname = True	×
178	urlopen(default_url + '/README.md')	×
179	except HTTPError: # e.g. 404	×
180	default_url = url_to_root + "raw/master"	×
181
182	warnings.warn(f"Data URL:{default_url}, error: {error}."	×
183	f"{error.reason}")
184
185	url = f"{getenv(env_var, default_url)}/{repo_path}"	5✔
186
187	return download(url, filepath, checksum, verbose)	5✔
188
189
190	def unzip(filepath, outdir=ELEPHANT_TMP_DIR, verbose=True):	5✔
191	with ZipFile(filepath) as zfile:	5✔
192	zfile.extractall(path=outdir)	5✔
193	if verbose:	5✔
194	print(f"Extracted {filepath} to {outdir}")	5✔

NeuralEnsemble / elephant / 11857958346

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous