17105429273

Committed 20 Aug 2025 05:13PM UTC coverage: 74.899% (-0.003%) from 74.902%

Build # 17105429273

Build Type

Pull #2696

github

Committed by

web-flow

Commit Message

Merge 87ce91d9b into 176901c06

Pull Request Pull Request #2696: discard marker fragments if size was zero

Run Details

5872 of 10620 branches covered (55.29%)

24194 of 32302 relevant lines covered (74.9%)

2.45 hits per line

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

88.18

/vispy/util/fetching.py

# -*- coding: utf-8 -*-
# Copyright (c) Vispy Development Team. All Rights Reserved.
# Distributed under the (new) BSD License. See LICENSE.txt for more info.

"""Data downloading and reading functions"""

from math import log
import os
from os import path as op
import sys
import shutil
import time

import urllib.request

from ..util.config import config


###############################################################################
# Vispy data directory

def load_data_file(fname, directory=None, force_download=False):
    """Get a standard vispy demo data file

    Parameters
    ----------
    fname : str
        The filename on the remote ``demo-data`` repository to download,
        e.g. ``'molecular_viewer/micelle.npy'``. These correspond to paths
        on ``https://github.com/vispy/demo-data/``.
    directory : str | None
        Directory to use to save the file. By default, the vispy
        configuration directory is used.
    force_download : bool | str
        If True, the file will be downloaded even if a local copy exists
        (and this copy will be overwritten). Can also be a YYYY-MM-DD date
        to ensure a file is up-to-date (modified date of a file on disk,
        if present, is checked).

    Returns
    -------
    fname : str
        The path to the file on the local system.
    """
    _url_root = 'https://raw.githubusercontent.com/vispy/demo-data/main/'
    url = _url_root + fname
    if directory is None:
        directory = config['data_path']
        if directory is None:
            raise ValueError('config["data_path"] is not defined, '
                             'so directory must be supplied')

    fname = op.join(directory, op.normcase(fname))  # convert to native
    if op.isfile(fname):
        if not force_download:  # we're done
            return fname
        if isinstance(force_download, str):
            ntime = time.strptime(force_download, '%Y-%m-%d')
            ftime = time.gmtime(op.getctime(fname))
            if ftime >= ntime:
                return fname
            else:
                print('File older than %s, updating...' % force_download)
    if not op.isdir(op.dirname(fname)):
        os.makedirs(op.abspath(op.dirname(fname)))
    # let's go get the file
    _fetch_file(url, fname)
    return fname


###############################################################################
# File downloading (most adapted from mne-python)


class ProgressBar(object):
    """Class for generating a command-line progressbar

    Parameters
    ----------
    max_value : int
        Maximum value of process (e.g. number of samples to process, bytes to
        download, etc.).
    initial_value : int
        Initial value of process, useful when resuming process from a specific
        value, defaults to 0.
    mesg : str
        Message to include at end of progress bar.
    max_chars : int
        Number of characters to use for progress bar (be sure to save some room
        for the message and % complete as well).
    progress_character : char
        Character in the progress bar that indicates the portion completed.
    spinner : bool
        Show a spinner.  Useful for long-running processes that may not
        increment the progress bar very often.  This provides the user with
        feedback that the progress has not stalled.
    """

    spinner_symbols = ['|', '/', '-', '\\']
    template = '\r[{0}{1}] {2:.05f} {3} {4}   '

    def __init__(self, max_value, initial_value=0, mesg='', max_chars=40,
                 progress_character='.', spinner=False):
        self.cur_value = initial_value
        self.max_value = float(max_value)
        self.mesg = mesg
        self.max_chars = max_chars
        self.progress_character = progress_character
        self.spinner = spinner
        self.spinner_index = 0
        self.n_spinner = len(self.spinner_symbols)

    def update(self, cur_value, mesg=None):
        """Update progressbar with current value of process

        Parameters
        ----------
        cur_value : number
            Current value of process.  Should be <= max_value (but this is not
            enforced).  The percent of the progressbar will be computed as
            (cur_value / max_value) * 100
        mesg : str
            Message to display to the right of the progressbar.  If None, the
            last message provided will be used.  To clear the current message,
            pass a null string, ''.
        """
        # Ensure floating-point division so we can get fractions of a percent
        # for the progressbar.
        self.cur_value = cur_value
        progress = float(self.cur_value) / self.max_value
        num_chars = int(progress * self.max_chars)
        num_left = self.max_chars - num_chars

        # Update the message
        if mesg is not None:
            self.mesg = mesg

        # The \r tells the cursor to return to the beginning of the line rather
        # than starting a new line.  This allows us to have a progressbar-style
        # display in the console window.
        bar = self.template.format(self.progress_character * num_chars,
                                   ' ' * num_left,
                                   progress * 100,
                                   self.spinner_symbols[self.spinner_index],
                                   self.mesg)
        sys.stdout.write(bar)
        # Increament the spinner
        if self.spinner:
            self.spinner_index = (self.spinner_index + 1) % self.n_spinner

        # Force a flush because sometimes when using bash scripts and pipes,
        # the output is not printed until after the program exits.
        sys.stdout.flush()

    def update_with_increment_value(self, increment_value, mesg=None):
        """Update progressbar with the value of the increment instead of the
        current value of process as in update()

        Parameters
        ----------
        increment_value : int
            Value of the increment of process.  The percent of the progressbar
            will be computed as
            (self.cur_value + increment_value / max_value) * 100
        mesg : str
            Message to display to the right of the progressbar.  If None, the
            last message provided will be used.  To clear the current message,
            pass a null string, ''.
        """
        self.cur_value += increment_value
        self.update(self.cur_value, mesg)


def _chunk_read(response, local_file, chunk_size=65536, initial_size=0):
    """Download a file chunk by chunk and show advancement

    Can also be used when resuming downloads over http.

    Parameters
    ----------
    response: urllib.response.addinfourl
        Response to the download request in order to get file size.
    local_file: file
        Hard disk file where data should be written.
    chunk_size: integer, optional
        Size of downloaded chunks. Default: 8192
    initial_size: int, optional
        If resuming, indicate the initial size of the file.
    """
    # Adapted from NISL:
    # https://github.com/nisl/tutorial/blob/master/nisl/datasets.py

    bytes_so_far = initial_size
    # Returns only amount left to download when resuming, not the size of the
    # entire file
    total_size = int(response.headers['Content-Length'].strip())
    total_size += initial_size

    progress = ProgressBar(total_size, initial_value=bytes_so_far,
                           max_chars=40, spinner=True, mesg='downloading')
    while True:
        chunk = response.read(chunk_size)
        bytes_so_far += len(chunk)
        if not chunk:
            sys.stderr.write('\n')
            break
        _chunk_write(chunk, local_file, progress)


def _chunk_write(chunk, local_file, progress):
    """Write a chunk to file and update the progress bar"""
    local_file.write(chunk)
    progress.update_with_increment_value(len(chunk))


def _fetch_file(url, file_name, print_destination=True):
    """Load requested file, downloading it if needed or requested

    Parameters
    ----------
    url: string
        The url of file to be downloaded.
    file_name: string
        Name, along with the path, of where downloaded file will be saved.
    print_destination: bool, optional
        If true, destination of where file was saved will be printed after
        download finishes.
    """
    # Adapted from NISL:
    # https://github.com/nisl/tutorial/blob/master/nisl/datasets.py

    temp_file_name = file_name + ".part"
    local_file = None
    initial_size = 0
    # Checking file size and displaying it alongside the download url
    n_try = 3
    for ii in range(n_try):
        try:
            data = urllib.request.urlopen(url, timeout=15.)
        except Exception as e:
            if ii == n_try - 1:
                raise RuntimeError('Error while fetching file %s.\n'
                                   'Dataset fetching aborted (%s)' % (url, e))
    try:
        file_size = int(data.headers['Content-Length'].strip())
        print('Downloading data from %s (%s)' % (url, sizeof_fmt(file_size)))
        local_file = open(temp_file_name, "wb")
        _chunk_read(data, local_file, initial_size=initial_size)
        # temp file must be closed prior to the move
        if not local_file.closed:
            local_file.close()
        shutil.move(temp_file_name, file_name)
        if print_destination is True:
            sys.stdout.write('File saved as %s.\n' % file_name)
    except Exception as e:
        raise RuntimeError('Error while fetching file %s.\n'
                           'Dataset fetching aborted (%s)' % (url, e))
    finally:
        if local_file is not None:
            if not local_file.closed:
                local_file.close()


def sizeof_fmt(num):
    """Turn number of bytes into human-readable str"""
    units = ['bytes', 'kB', 'MB', 'GB', 'TB', 'PB']
    decimals = [0, 0, 1, 2, 2, 2]
    """Human friendly file size"""
    if num > 1:
        exponent = min(int(log(num, 1024)), len(units) - 1)
        quotient = float(num) / 1024 ** exponent
        unit = units[exponent]
        num_decimals = decimals[exponent]
        format_string = '{0:.%sf} {1}' % (num_decimals)
        return format_string.format(quotient, unit)
    return '0 bytes' if num == 0 else '1 byte'

1	# -- coding: utf-8 --
2	# Copyright (c) Vispy Development Team. All Rights Reserved.
3	# Distributed under the (new) BSD License. See LICENSE.txt for more info.
4
5	"""Data downloading and reading functions"""	4✔
6
7	from math import log	4✔
8	import os	4✔
9	from os import path as op	4✔
10	import sys	4✔
11	import shutil	4✔
12	import time	4✔
13
14	import urllib.request	4✔
15
16	from ..util.config import config	4✔
17
18
19	###############################################################################
20	# Vispy data directory
21
22	def load_data_file(fname, directory=None, force_download=False):	4✔
23	"""Get a standard vispy demo data file
24
25	Parameters
26	----------
27	fname : str
28	The filename on the remote ``demo-data`` repository to download,
29	e.g. ``'molecular_viewer/micelle.npy'``. These correspond to paths
30	on ``https://github.com/vispy/demo-data/``.
31	directory : str \| None
32	Directory to use to save the file. By default, the vispy
33	configuration directory is used.
34	force_download : bool \| str
35	If True, the file will be downloaded even if a local copy exists
36	(and this copy will be overwritten). Can also be a YYYY-MM-DD date
37	to ensure a file is up-to-date (modified date of a file on disk,
38	if present, is checked).
39
40	Returns
41	-------
42	fname : str
43	The path to the file on the local system.
44	"""
45	_url_root = 'https://raw.githubusercontent.com/vispy/demo-data/main/'	4✔
46	url = _url_root + fname	4✔
47	if directory is None:	4!
48	directory = config['data_path']	4✔
49	if directory is None:	4!
50	raise ValueError('config["data_path"] is not defined, '	×
51	'so directory must be supplied')
52
53	fname = op.join(directory, op.normcase(fname)) # convert to native	4✔
54	if op.isfile(fname):	4✔
55	if not force_download: # we're done	4!
56	return fname	4✔
57	if isinstance(force_download, str):	×
58	ntime = time.strptime(force_download, '%Y-%m-%d')	×
59	ftime = time.gmtime(op.getctime(fname))	×
60	if ftime >= ntime:	×
61	return fname	×
62	else:
63	print('File older than %s, updating...' % force_download)	×
64	if not op.isdir(op.dirname(fname)):	3!
65	os.makedirs(op.abspath(op.dirname(fname)))	×
66	# let's go get the file
67	_fetch_file(url, fname)	3✔
68	return fname	3✔
69
70
71	###############################################################################
72	# File downloading (most adapted from mne-python)
73
74
75	class ProgressBar(object):	4✔
76	"""Class for generating a command-line progressbar
77
78	Parameters
79	----------
80	max_value : int
81	Maximum value of process (e.g. number of samples to process, bytes to
82	download, etc.).
83	initial_value : int
84	Initial value of process, useful when resuming process from a specific
85	value, defaults to 0.
86	mesg : str
87	Message to include at end of progress bar.
88	max_chars : int
89	Number of characters to use for progress bar (be sure to save some room
90	for the message and % complete as well).
91	progress_character : char
92	Character in the progress bar that indicates the portion completed.
93	spinner : bool
94	Show a spinner. Useful for long-running processes that may not
95	increment the progress bar very often. This provides the user with
96	feedback that the progress has not stalled.
97	"""
98
99	spinner_symbols = ['\|', '/', '-', '\\']	4✔
100	template = '\r[{0}{1}] {2:.05f} {3} {4} '	4✔
101
102	def __init__(self, max_value, initial_value=0, mesg='', max_chars=40,	4✔
103	progress_character='.', spinner=False):
104	self.cur_value = initial_value	3✔
105	self.max_value = float(max_value)	3✔
106	self.mesg = mesg	3✔
107	self.max_chars = max_chars	3✔
108	self.progress_character = progress_character	3✔
109	self.spinner = spinner	3✔
110	self.spinner_index = 0	3✔
111	self.n_spinner = len(self.spinner_symbols)	3✔
112
113	def update(self, cur_value, mesg=None):	4✔
114	"""Update progressbar with current value of process
115
116	Parameters
117	----------
118	cur_value : number
119	Current value of process. Should be <= max_value (but this is not
120	enforced). The percent of the progressbar will be computed as
121	(cur_value / max_value) * 100
122	mesg : str
123	Message to display to the right of the progressbar. If None, the
124	last message provided will be used. To clear the current message,
125	pass a null string, ''.
126	"""
127	# Ensure floating-point division so we can get fractions of a percent
128	# for the progressbar.
129	self.cur_value = cur_value	3✔
130	progress = float(self.cur_value) / self.max_value	3✔
131	num_chars = int(progress * self.max_chars)	3✔
132	num_left = self.max_chars - num_chars	3✔
133
134	# Update the message
135	if mesg is not None:	3!
136	self.mesg = mesg	×
137
138	# The \r tells the cursor to return to the beginning of the line rather
139	# than starting a new line. This allows us to have a progressbar-style
140	# display in the console window.
141	bar = self.template.format(self.progress_character * num_chars,	3✔
142	' ' * num_left,
143	progress * 100,
144	self.spinner_symbols[self.spinner_index],
145	self.mesg)
146	sys.stdout.write(bar)	3✔
147	# Increament the spinner
148	if self.spinner:	3!
149	self.spinner_index = (self.spinner_index + 1) % self.n_spinner	3✔
150
151	# Force a flush because sometimes when using bash scripts and pipes,
152	# the output is not printed until after the program exits.
153	sys.stdout.flush()	3✔
154
155	def update_with_increment_value(self, increment_value, mesg=None):	4✔
156	"""Update progressbar with the value of the increment instead of the
157	current value of process as in update()
158
159	Parameters
160	----------
161	increment_value : int
162	Value of the increment of process. The percent of the progressbar
163	will be computed as
164	(self.cur_value + increment_value / max_value) * 100
165	mesg : str
166	Message to display to the right of the progressbar. If None, the
167	last message provided will be used. To clear the current message,
168	pass a null string, ''.
169	"""
170	self.cur_value += increment_value	3✔
171	self.update(self.cur_value, mesg)	3✔
172
173
174	def _chunk_read(response, local_file, chunk_size=65536, initial_size=0):	4✔
175	"""Download a file chunk by chunk and show advancement
176
177	Can also be used when resuming downloads over http.
178
179	Parameters
180	----------
181	response: urllib.response.addinfourl
182	Response to the download request in order to get file size.
183	local_file: file
184	Hard disk file where data should be written.
185	chunk_size: integer, optional
186	Size of downloaded chunks. Default: 8192
187	initial_size: int, optional
188	If resuming, indicate the initial size of the file.
189	"""
190	# Adapted from NISL:
191	# https://github.com/nisl/tutorial/blob/master/nisl/datasets.py
192
193	bytes_so_far = initial_size	3✔
194	# Returns only amount left to download when resuming, not the size of the
195	# entire file
196	total_size = int(response.headers['Content-Length'].strip())	3✔
197	total_size += initial_size	3✔
198
199	progress = ProgressBar(total_size, initial_value=bytes_so_far,	3✔
200	max_chars=40, spinner=True, mesg='downloading')
201	while True:	1✔
202	chunk = response.read(chunk_size)	3✔
203	bytes_so_far += len(chunk)	3✔
204	if not chunk:	3✔
205	sys.stderr.write('\n')	3✔
206	break	3✔
207	_chunk_write(chunk, local_file, progress)	3✔
208
209
210	def _chunk_write(chunk, local_file, progress):	4✔
211	"""Write a chunk to file and update the progress bar"""
212	local_file.write(chunk)	3✔
213	progress.update_with_increment_value(len(chunk))	3✔
214
215
216	def _fetch_file(url, file_name, print_destination=True):	4✔
217	"""Load requested file, downloading it if needed or requested
218
219	Parameters
220	----------
221	url: string
222	The url of file to be downloaded.
223	file_name: string
224	Name, along with the path, of where downloaded file will be saved.
225	print_destination: bool, optional
226	If true, destination of where file was saved will be printed after
227	download finishes.
228	"""
229	# Adapted from NISL:
230	# https://github.com/nisl/tutorial/blob/master/nisl/datasets.py
231
232	temp_file_name = file_name + ".part"	3✔
233	local_file = None	3✔
234	initial_size = 0	3✔
235	# Checking file size and displaying it alongside the download url
236	n_try = 3	3✔
237	for ii in range(n_try):	3✔
238	try:	3✔
239	data = urllib.request.urlopen(url, timeout=15.)	3✔
240	except Exception as e:	3✔
241	if ii == n_try - 1:	3✔
242	raise RuntimeError('Error while fetching file %s.\n'	3✔
243	'Dataset fetching aborted (%s)' % (url, e))
244	try:	3✔
245	file_size = int(data.headers['Content-Length'].strip())	3✔
246	print('Downloading data from %s (%s)' % (url, sizeof_fmt(file_size)))	3✔
247	local_file = open(temp_file_name, "wb")	3✔
248	_chunk_read(data, local_file, initial_size=initial_size)	3✔
249	# temp file must be closed prior to the move
250	if not local_file.closed:	3!
251	local_file.close()	3✔
252	shutil.move(temp_file_name, file_name)	3✔
253	if print_destination is True:	3!
254	sys.stdout.write('File saved as %s.\n' % file_name)	3✔
255	except Exception as e:	×
256	raise RuntimeError('Error while fetching file %s.\n'	×
257	'Dataset fetching aborted (%s)' % (url, e))
258	finally:
259	if local_file is not None:	3!
260	if not local_file.closed:	3!
261	local_file.close()	×
262
263
264	def sizeof_fmt(num):	4✔
265	"""Turn number of bytes into human-readable str"""
266	units = ['bytes', 'kB', 'MB', 'GB', 'TB', 'PB']	3✔
267	decimals = [0, 0, 1, 2, 2, 2]	3✔
268	"""Human friendly file size"""	1✔
269	if num > 1:	3!
270	exponent = min(int(log(num, 1024)), len(units) - 1)	3✔
271	quotient = float(num) / 1024 ** exponent	3✔
272	unit = units[exponent]	3✔
273	num_decimals = decimals[exponent]	3✔
274	format_string = '{0:.%sf} {1}' % (num_decimals)	3✔
275	return format_string.format(quotient, unit)	3✔
276	return '0 bytes' if num == 0 else '1 byte'	×

vispy / vispy / 17105429273

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