11308274429

Committed 12 Oct 2024 07:37PM CUT coverage: 96.177% (-0.005%) from 96.182%

Build # 11308274429

Build Type

push

github

Committed by

KelSolaar

Commit Message

Merge branch 'release/v0.2.6'

Run Details

238 of 241 new or added lines in 43 files covered. (98.76%)

2 existing lines in 2 files now uncovered.

2214 of 2302 relevant lines covered (96.18%)

0.96 hits per line

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

90.53

/colour_datasets/utilities/common.py

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

Define 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 socket
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:
                timeout = socket.getdefaulttimeout()
                try:
                    socket.setdefaulttimeout(10)
                    urllib.request.urlretrieve(  # noqa: S310
                        url,
                        filename=filename,
                        reporthook=progress.update_to,
                        data=None,
                    )
                finally:
                    socket.setdefaulttimeout(timeout)

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

colour-science / colour-datasets / 11308274429

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