7853010122

Committed 10 Feb 2024 05:53AM CUT coverage: 96.225%. Remained the same

Build # 7853010122

Build Type

push

github

Committed by

KelSolaar

Commit Message

Use `np.reshape` function instead of `.reshape` method.

Run Details

2243 of 2331 relevant lines covered (96.22%)

0.96 hits per line

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

90.0

/colour_datasets/utilities/common.py

"""
Common Utilities
================

Defines the common utilities objects that don't fall in any specific category.
"""

from __future__ import annotations

import functools
import gzip
import hashlib
import json
import os
import shutil
import sys
import urllib.error
import urllib.request

import setuptools.archive_util
from cachetools import TTLCache, cached
from colour.hints import Any, Callable, Dict
from tqdm import tqdm

__author__ = "Colour Developers"
__copyright__ = "Copyright 2019 Colour Developers"
__license__ = "BSD-3-Clause - https://opensource.org/licenses/BSD-3-Clause"
__maintainer__ = "Colour Developers"
__email__ = "colour-developers@colour-science.org"
__status__ = "Production"

__all__ = [
    "suppress_stdout",
    "TqdmUpTo",
    "hash_md5",
    "url_download",
    "json_open",
    "unpack_gzipfile",
]


# TODO: Use *colour* definition.
class suppress_stdout:
    """A context manager and decorator temporarily suppressing standard output."""

    def __enter__(self) -> suppress_stdout:
        """Redirect the standard output upon entering the context manager."""

        self._stdout = sys.stdout
        sys.stdout = open(os.devnull, "w")  # noqa: SIM115

        return self

    def __exit__(self, *args: Any):
        """Restore the standard output upon exiting the context manager."""

        sys.stdout.close()
        sys.stdout = self._stdout

    def __call__(self, function: Callable) -> Callable:
        """Call the wrapped definition."""

        @functools.wraps(function)
        def wrapper(*args: Any, **kwargs: Any) -> Callable:
            with self:
                return function(*args, **kwargs)

        return wrapper


class TqdmUpTo(tqdm):
    """:class:`tqdm` sub-class used to report the progress of an action."""

    def update_to(
        self,
        chunks_count: int = 1,
        chunk_size: int = 1,
        total_size: int | None = None,
    ):
        """
        Report the progress of an action.

        Parameters
        ----------
        chunks_count
            Number of blocks transferred.
        chunk_size
            Size of each block (in tqdm units).
        total_size
            Total size (in tqdm units).
        """

        if total_size is not None:
            self.total = total_size

        self.update(chunks_count * chunk_size - self.n)


def hash_md5(filename: str, chunk_size: int = 2**16) -> str:
    """
    Compute the *Message Digest 5 (MD5)* hash of given file.

    Parameters
    ----------
    filename
        File to compute the *MD5* hash of.
    chunk_size
        Chunk size to read from the file.

    Returns
    -------
    :class:`str`
        *MD5* hash of given file.
    """

    md5 = hashlib.md5()  # noqa: S324

    with open(filename, "rb") as file_object:
        while True:
            chunk = file_object.read(chunk_size)
            if not chunk:
                break

            md5.update(chunk)

    return md5.hexdigest()


def url_download(url: str, filename: str, md5: str | None = None, retries: int = 3):
    """
    Download given url and saves its content at given file.

    Parameters
    ----------
    url
        Url to download.
    filename
        File to save the url content at.
    md5
        *Message Digest 5 (MD5)* hash of the content at given url. If provided
        the saved content at given file will be hashed and compared to ``md5``.
    retries
        Number of retries in case where a networking error occurs or the *MD5*
        hash is not matching.

    Examples
    --------
    >>> import os
    >>> url_download("https://github.com/colour-science/colour-datasets", os.devnull)
    """

    attempt = 0
    while attempt != retries:
        try:
            with TqdmUpTo(
                unit="B",
                unit_scale=True,
                miniters=1,
                desc=f'Downloading "{url}" url',
            ) as progress:
                urllib.request.urlretrieve(  # noqa: S310
                    url,
                    filename=filename,
                    reporthook=progress.update_to,
                    data=None,
                )

            if md5 is not None and md5.lower() != hash_md5(filename):
                raise ValueError(  # noqa: TRY301
                    f'"MD5" hash of "{filename}" file does not match the '
                    f"expected hash!"
                )

            attempt = retries
        except (urllib.error.URLError, OSError, ValueError):
            attempt += 1
            print(  # noqa: T201
                f'An error occurred while downloading "{filename}" file '
                f"during attempt {attempt}, retrying..."
            )
            if attempt == retries:
                raise


@cached(cache=TTLCache(maxsize=256, ttl=300))
def json_open(url: str, retries: int = 3) -> Dict:
    """
    Open given url and return its content as *JSON*.

    Parameters
    ----------
    url
        Url to open.
    retries
        Number of retries in case where a networking error occurs.

    Returns
    -------
    :class:`dict`
        *JSON* data.

    Raises
    ------
    urllib.error.URLError, ValueError
        If the url cannot be opened or parsed as *JSON*.

    Notes
    -----
    -   The definition caches the request *JSON* output for 5 minutes.

    Examples
    --------
    >>> json_open("https://zenodo.org/api/records/3245883")
    ... # doctest: +SKIP
    '{"conceptdoi":"10.5281/zenodo.3245882"'
    """

    data: Dict = {}

    attempt = 0
    while attempt != retries:
        try:
            request = urllib.request.Request(url)  # noqa: S310
            with urllib.request.urlopen(request) as response:  # noqa: S310
                return json.loads(response.read())
        except (urllib.error.URLError, ValueError):
            attempt += 1
            print(  # noqa: T201
                f'An error occurred while opening "{url}" url during attempt '
                f"{attempt}, retrying..."
            )
            if attempt == retries:
                raise

    return data


def unpack_gzipfile(
    filename: str,
    extraction_directory: str,
    *args: Any,  # noqa: ARG001
) -> bool:
    """
    Unpack given *GZIP* file to given extraction directory.

    Parameters
    ----------
    filename
        *GZIP* file to extract.
    extraction_directory
        Directory where to extract the *GZIP* file.

    Other Parameters
    ----------------
    args
        Arguments.

    Returns
    -------
    :class:`bool`
        Definition success.

    Notes
    -----
    -   This definition is used as an extra driver for
        :func:`setuptools.archive_util.unpack archive` definition.
    """

    extraction_path = os.path.join(
        extraction_directory, os.path.splitext(os.path.basename(filename))[0]
    )

    if not os.path.exists(extraction_directory):
        os.makedirs(extraction_directory)

    try:
        with gzip.open(filename) as gzip_file, open(
            extraction_path, "wb"
        ) as output_file:
            shutil.copyfileobj(gzip_file, output_file)
    except Exception as error:
        print(error)  # noqa: T201
        raise setuptools.archive_util.UnrecognizedFormat(
            f'{filename} is not a "GZIP" file!'
        ) from error

    return True


setuptools.archive_util.extraction_drivers = (
    setuptools.archive_util.unpack_directory,
    setuptools.archive_util.unpack_zipfile,
    setuptools.archive_util.unpack_tarfile,
    unpack_gzipfile,
)

1	"""
2	Common Utilities
3	================
4
5	Defines the common utilities objects that don't fall in any specific category.
6	"""
7
8	from __future__ import annotations	1✔
9
10	import functools	1✔
11	import gzip	1✔
12	import hashlib	1✔
13	import json	1✔
14	import os	1✔
15	import shutil	1✔
16	import sys	1✔
17	import urllib.error	1✔
18	import urllib.request	1✔
19
20	import setuptools.archive_util	1✔
21	from cachetools import TTLCache, cached	1✔
22	from colour.hints import Any, Callable, Dict	1✔
23	from tqdm import tqdm	1✔
24
25	__author__ = "Colour Developers"	1✔
26	__copyright__ = "Copyright 2019 Colour Developers"	1✔
27	__license__ = "BSD-3-Clause - https://opensource.org/licenses/BSD-3-Clause"	1✔
28	__maintainer__ = "Colour Developers"	1✔
29	__email__ = "colour-developers@colour-science.org"	1✔
30	__status__ = "Production"	1✔
31
32	__all__ = [	1✔
33	"suppress_stdout",
34	"TqdmUpTo",
35	"hash_md5",
36	"url_download",
37	"json_open",
38	"unpack_gzipfile",
39	]
40
41
42	# TODO: Use colour definition.
43	class suppress_stdout:	1✔
44	"""A context manager and decorator temporarily suppressing standard output."""
45
46	def __enter__(self) -> suppress_stdout:	1✔
47	"""Redirect the standard output upon entering the context manager."""
48
49	self._stdout = sys.stdout	1✔
50	sys.stdout = open(os.devnull, "w") # noqa: SIM115	1✔
51
52	return self	1✔
53
54	def __exit__(self, *args: Any):	1✔
55	"""Restore the standard output upon exiting the context manager."""
56
57	sys.stdout.close()	1✔
58	sys.stdout = self._stdout	1✔
59
60	def __call__(self, function: Callable) -> Callable:	1✔
61	"""Call the wrapped definition."""
62
63	@functools.wraps(function)	×
64	def wrapper(args: Any, *kwargs: Any) -> Callable:	×
65	with self:	×
66	return function(args, *kwargs)	×
67
68	return wrapper	×
69
70
71	class TqdmUpTo(tqdm):	1✔
72	""":class:`tqdm` sub-class used to report the progress of an action."""
73
74	def update_to(	1✔
75	self,
76	chunks_count: int = 1,
77	chunk_size: int = 1,
78	total_size: int \| None = None,
79	):
80	"""
81	Report the progress of an action.
82
83	Parameters
84	----------
85	chunks_count
86	Number of blocks transferred.
87	chunk_size
88	Size of each block (in tqdm units).
89	total_size
90	Total size (in tqdm units).
91	"""
92
93	if total_size is not None:	1✔
94	self.total = total_size	1✔
95
96	self.update(chunks_count * chunk_size - self.n)	1✔
97
98
99	def hash_md5(filename: str, chunk_size: int = 2**16) -> str:	1✔
100	"""
101	Compute the Message Digest 5 (MD5) hash of given file.
102
103	Parameters
104	----------
105	filename
106	File to compute the MD5 hash of.
107	chunk_size
108	Chunk size to read from the file.
109
110	Returns
111	-------
112	:class:`str`
113	MD5 hash of given file.
114	"""
115
116	md5 = hashlib.md5() # noqa: S324	1✔
117
118	with open(filename, "rb") as file_object:	1✔
119	while True:	1✔
120	chunk = file_object.read(chunk_size)	1✔
121	if not chunk:	1✔
122	break	1✔
123
124	md5.update(chunk)	1✔
125
126	return md5.hexdigest()	1✔
127
128
129	def url_download(url: str, filename: str, md5: str \| None = None, retries: int = 3):	1✔
130	"""
131	Download given url and saves its content at given file.
132
133	Parameters
134	----------
135	url
136	Url to download.
137	filename
138	File to save the url content at.
139	md5
140	Message Digest 5 (MD5) hash of the content at given url. If provided
141	the saved content at given file will be hashed and compared to ``md5``.
142	retries
143	Number of retries in case where a networking error occurs or the MD5
144	hash is not matching.
145
146	Examples
147	--------
148	>>> import os
149	>>> url_download("https://github.com/colour-science/colour-datasets", os.devnull)
150	"""
151
152	attempt = 0	1✔
153	while attempt != retries:	1✔
154	try:	1✔
155	with TqdmUpTo(	1✔
156	unit="B",
157	unit_scale=True,
158	miniters=1,
159	desc=f'Downloading "{url}" url',
160	) as progress:
161	urllib.request.urlretrieve( # noqa: S310	1✔
162	url,
163	filename=filename,
164	reporthook=progress.update_to,
165	data=None,
166	)
167
168	if md5 is not None and md5.lower() != hash_md5(filename):	1✔
169	raise ValueError( # noqa: TRY301	1✔
170	f'"MD5" hash of "{filename}" file does not match the '
171	f"expected hash!"
172	)
173
174	attempt = retries	1✔
175	except (urllib.error.URLError, OSError, ValueError):	1✔
176	attempt += 1	1✔
177	print( # noqa: T201	1✔
178	f'An error occurred while downloading "{filename}" file '
179	f"during attempt {attempt}, retrying..."
180	)
181	if attempt == retries:	1✔
182	raise	1✔
183
184
185	@cached(cache=TTLCache(maxsize=256, ttl=300))	1✔
186	def json_open(url: str, retries: int = 3) -> Dict:	1✔
187	"""
188	Open given url and return its content as JSON.
189
190	Parameters
191	----------
192	url
193	Url to open.
194	retries
195	Number of retries in case where a networking error occurs.
196
197	Returns
198	-------
199	:class:`dict`
200	JSON data.
201
202	Raises
203	------
204	urllib.error.URLError, ValueError
205	If the url cannot be opened or parsed as JSON.
206
207	Notes
208	-----
209	- The definition caches the request JSON output for 5 minutes.
210
211	Examples
212	--------
213	>>> json_open("https://zenodo.org/api/records/3245883")
214	... # doctest: +SKIP
215	'{"conceptdoi":"10.5281/zenodo.3245882"'
216	"""
217
218	data: Dict = {}	1✔
219
220	attempt = 0	1✔
221	while attempt != retries:	1✔
222	try:	1✔
223	request = urllib.request.Request(url) # noqa: S310	1✔
224	with urllib.request.urlopen(request) as response: # noqa: S310	1✔
225	return json.loads(response.read())	1✔
226	except (urllib.error.URLError, ValueError):	1✔
227	attempt += 1	1✔
228	print( # noqa: T201	1✔
229	f'An error occurred while opening "{url}" url during attempt '
230	f"{attempt}, retrying..."
231	)
232	if attempt == retries:	1✔
233	raise	1✔
234
235	return data	×
236
237
238	def unpack_gzipfile(	1✔
239	filename: str,
240	extraction_directory: str,
241	*args: Any, # noqa: ARG001
242	) -> bool:
243	"""
244	Unpack given GZIP file to given extraction directory.
245
246	Parameters
247	----------
248	filename
249	GZIP file to extract.
250	extraction_directory
251	Directory where to extract the GZIP file.
252
253	Other Parameters
254	----------------
255	args
256	Arguments.
257
258	Returns
259	-------
260	:class:`bool`
261	Definition success.
262
263	Notes
264	-----
265	- This definition is used as an extra driver for
266	:func:`setuptools.archive_util.unpack archive` definition.
267	"""
268
269	extraction_path = os.path.join(	1✔
270	extraction_directory, os.path.splitext(os.path.basename(filename))[0]
271	)
272
273	if not os.path.exists(extraction_directory):	1✔
274	os.makedirs(extraction_directory)	1✔
275
276	try:	1✔
277	with gzip.open(filename) as gzip_file, open(	1✔
278	extraction_path, "wb"
279	) as output_file:
280	shutil.copyfileobj(gzip_file, output_file)	1✔
281	except Exception as error:	×
282	print(error) # noqa: T201	×
283	raise setuptools.archive_util.UnrecognizedFormat(	×
284	f'{filename} is not a "GZIP" file!'
285	) from error
286
287	return True	1✔
288
289
290	setuptools.archive_util.extraction_drivers = (	1✔
291	setuptools.archive_util.unpack_directory,
292	setuptools.archive_util.unpack_zipfile,
293	setuptools.archive_util.unpack_tarfile,
294	unpack_gzipfile,
295	)

colour-science / colour-datasets / 7853010122

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