import io import logging import os import platform import re import threading import urllib.parse import urllib.request from wfdb.version import __version__ # Value for 'buffering' indicating that the entire file should be # buffered at once. BUFFER_WHOLE_FILE = -2 # Default buffer size for remote files. DEFAULT_BUFFER_SIZE = 32768 # Logger for this module. _LOGGER = logging.getLogger(__name__) # Pattern that matches the value of the Content-Range response header. _CONTENT_RANGE_PATTERN = re.compile( r"bytes (?:(\d+)-(\d+)|\*)/(?:(\d+)|\*)", re.ASCII | re.IGNORECASE ) # Global session object. _SESSION = None _SESSION_PID = None _SESSION_LOCK = threading.Lock() def _get_session(): """ Obtain a session object suitable for requesting remote files. Parameters ---------- N/A Returns ------- session : requests.Session A session object. """ import requests import requests.adapters global _SESSION global _SESSION_PID with _SESSION_LOCK: if _SESSION is None: _SESSION = requests.Session() _SESSION.headers["User-Agent"] = " ".join( [ "%s/%s" % ("wfdb-python", __version__), "%s/%s" % ("python-requests", requests.__version__), "%s/%s" % ( platform.python_implementation(), platform.python_version(), ), ] ) for protocol in ("http", "https"): adapter = requests.adapters.HTTPAdapter( pool_maxsize=2, pool_block=True, ) _SESSION.mount("%s://" % protocol, adapter) # Ensure we don't reuse sockets after forking if _SESSION_PID != os.getpid(): _SESSION_PID = os.getpid() _SESSION.close() return _SESSION class NetFileError(OSError): """An error occurred while reading a remote file.""" def __init__(self, message, url=None, status_code=None): super().__init__(message) self.url = url self.status_code = status_code class NetFileNotFoundError(NetFileError, FileNotFoundError): """A remote file does not exist.""" class NetFilePermissionError(NetFileError, PermissionError): """The client does not have permission to access a remote file.""" class RangeTransfer: """ A single HTTP transfer representing a range of bytes. Parameters ---------- url : str URL of the remote file. start : int, optional Start of the byte range to download, as an offset from the beginning of the file (inclusive, 0-based.) end : int or None End of the byte range to download, as an offset from the beginning of the file (exclusive, 0-based.) If None, request all data until the end of the file. Attributes ---------- request_url : str Original URL that was requested. response_url : str URL that was actually retrieved (after following redirections.) is_complete : bool True if the response contains the entire file; False if the response contains a byte range. file_size : int or None Total size of the remote file. This may be None if the length is unknown. Notes ----- The start and end parameters are requests that the server may or may not honor. After creating a RangeTransfer object, call content() or iter_chunks() to retrieve the actual response data, which may be a subset or a superset of the requested range. """ def __init__(self, url, start, end): self.request_url = url if start == 0 and end is None: method = "GET" headers = {} elif end is None: method = "GET" headers = { "Range": "bytes=%d-" % start, "Accept-Encoding": None, } elif end > start: method = "GET" headers = { "Range": "bytes=%d-%d" % (start, end - 1), "Accept-Encoding": None, } else: method = "HEAD" headers = { "Accept-Encoding": None, } session = _get_session() self._response = session.request( method, url, headers=headers, stream=True ) self._content_iter = self._response.iter_content(4096) try: self._parse_headers(method, self._response) except Exception: self.close() raise def _parse_headers(self, method, response): """ Parse the headers of the response object. Parameters ---------- method : str The HTTP method used for the request. response : requests.Response The resulting response object. Returns ------- N/A Notes ----- - response_url is set to the URL of the response - file_size is set to the total file size - is_complete is set to true if the response is complete - _current_pos is set to the starting position - _expected_end_pos is set to the expected end position """ self.response_url = response.url self.file_size = None self.is_complete = False self._current_pos = 0 self._expected_end_pos = None # Raise an exception if an error occurs. if response.status_code >= 400 and response.status_code != 416: _LOGGER.info( "%s %s: %s", method, response.url, response.status_code ) if response.status_code in (401, 403): cls = NetFilePermissionError elif response.status_code == 404: cls = NetFileNotFoundError else: cls = NetFileError raise cls( "%s Error: %s for url: %s" % (response.status_code, response.reason, response.url), url=response.url, status_code=response.status_code, ) # Parse the Content-Range if this is a partial response. elif response.status_code in (206, 416): content_range = response.headers.get("Content-Range") if content_range: match = _CONTENT_RANGE_PATTERN.fullmatch(content_range) if not match: raise NetFileError( "Invalid Content-Range: %s" % content_range, url=response.url, ) if match.group(1): self._current_pos = int(match.group(1)) self._expected_end_pos = int(match.group(2)) + 1 if match.group(3): self.file_size = int(match.group(3)) elif response.status_code == 206: raise NetFileError( "Missing Content-Range in partial response", url=response.url, ) # Parse the Content-Length if this is a complete and # uncompressed response. elif 200 <= response.status_code < 300: self.is_complete = True content_encoding = response.headers.get("Content-Encoding") content_length = response.headers.get("Content-Length") if content_length and not content_encoding: try: self.file_size = int(content_length) self._expected_end_pos = self.file_size except ValueError: raise NetFileError( "Invalid Content-Length: %s" % content_length, url=response.url, ) _LOGGER.info( "%s %s: %s %s-%s/%s", method, response.url, response.status_code, self._current_pos, self._expected_end_pos, self.file_size, ) # If the response is an error (or an unhandled redirection) # then discard the body. if response.status_code >= 300: self.close() def close(self): """ Finish reading data from the response. Any leftover data in the response body will be discarded and the underlying HTTP connection will be returned to the pool. Parameters ---------- N/A Returns ------- N/A """ try: for data in self._content_iter: pass except Exception: pass self._response.close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): # When exiting with a normal exception, shut down cleanly by # reading leftover response data. When exiting abnormally # (SystemExit, KeyboardInterrupt), do nothing. if not exc_type or issubclass(exc_type, Exception): self.close() def __del__(self): # If the object is deleted without calling close(), forcibly # close the existing connection. response = getattr(self, "_response", None) if response: response.close() def iter_chunks(self): """ Iterate over the response body as a sequence of chunks. Parameters ---------- N/A Yields ------ chunk_start : int Byte offset within the remote file corresponding to the start of this chunk. chunk_data : bytes Contents of the chunk. """ for chunk_data in self._content_iter: chunk_start = self._current_pos self._current_pos += len(chunk_data) yield chunk_start, chunk_data if self.is_complete: self.file_size = self._current_pos def content(self): """ Read the complete response. Parameters ---------- N/A Returns ------- start : int Byte offset within the remote file corresponding to the start of the response. data : bytes Contents of the response. """ start = self._current_pos chunks = [] for _, chunk_data in self.iter_chunks(): chunks.append(chunk_data) return start, b"".join(chunks) class NetFile(io.BufferedIOBase): """ File object providing random access to a remote file over HTTP. Attributes ---------- url : str URL of the remote file. buffering : int, optional Buffering policy. If buffering = 0, internal buffering is disabled; each operation on the stream requires a separate request to the server. If buffering = -2, the entire file is downloaded in a single request. If buffering > 0, it specifies the minimum size of the internal buffer. If buffering = -1, the default buffer size is used. """ def __init__(self, url, buffering=-1): self.url = url self.name = url self.buffering = buffering self._pos = 0 self._file_size = None self._buffer = b"" self._buffer_start = 0 self._buffer_end = 0 self._current_url = self.url def _read_buffered_range(self, start, end): """ Read a range of bytes from the internal buffer. Parameters ---------- start : int Starting byte offset of the desired range. end : int Ending byte offset of the desired range. Returns ------- data : memoryview A memoryview of the given byte range. """ bstart = start - self._buffer_start bend = end - self._buffer_start if 0 <= bstart <= bend: return memoryview(self._buffer)[bstart:bend] else: return memoryview(b"") def _read_range(self, start, end): """ Read a range of bytes from the remote file. The result is returned as a sequence of chunks; the sizes of the individual chunks are unspecified. The total size may be less than requested if the end of the file is reached. Parameters ---------- start : int Starting byte offset of the desired range. end : int or None Ending byte offset of the desired range, or None to read all data up to the end of the file. Yields ------ data : memoryview A memoryview containing a chunk of the desired range. """ # Read buffered data if available. if self._buffer_start <= start < self._buffer_end: if end is None: range_end = self._buffer_end else: range_end = min(end, self._buffer_end) yield self._read_buffered_range(start, range_end) start = range_end if end is not None and start >= end: return if self._file_size is not None and start >= self._file_size: return buffer_store = False if self.buffering == BUFFER_WHOLE_FILE: # Request entire file and save it in the internal buffer. req_start = 0 req_end = None buffer_store = True elif end is None: # Request range from start to EOF and don't save it in the # buffer (since the result will be immediately consumed.) req_start = start req_end = None else: # Request a fixed range of bytes. Save it in the buffer # if it is smaller than the maximum buffer size. buffer_size = self.buffering if buffer_size < 0: buffer_size = DEFAULT_BUFFER_SIZE req_start = start req_end = end if req_end < req_start + buffer_size: req_end = req_start + buffer_size buffer_store = True with RangeTransfer(self._current_url, req_start, req_end) as xfer: # Update current file URL. self._current_url = xfer.response_url # If we requested a range but the server doesn't support # random access, then unless buffering is disabled, save # entire file in the buffer. if self.buffering == 0: buffer_store = False elif xfer.is_complete and (start, end) != (0, None): buffer_store = True if buffer_store: # Load data into buffer and then return a copy to the # caller. (start, data) = xfer.content() self._buffer = data self._buffer_start = start self._buffer_end = start + len(data) if end is None: end = self._buffer_end yield self._read_buffered_range(start, end) else: # Return requested data to caller without buffering. for chunk_start, chunk_data in xfer.iter_chunks(): rel_start = start - chunk_start if 0 <= rel_start < len(chunk_data): if end is None: rel_end = len(chunk_data) else: rel_end = min(end - chunk_start, len(chunk_data)) yield memoryview(chunk_data)[rel_start:rel_end] start = chunk_start + rel_end # Update file size. if self.buffering != 0: self._file_size = xfer.file_size def _get_size(self): """ Determine the size of the remote file. Parameters ---------- N/A Returns ------- size : int or None Size of the remote file, if known. """ size = self._file_size if size is None: if self.buffering == BUFFER_WHOLE_FILE: for _ in self._read_range(0, None): pass else: with RangeTransfer(self._current_url, 0, 0) as xfer: self._current_url = xfer.response_url self._file_size = xfer.file_size size = self._file_size if self.buffering == 0: self._file_size = None return size def readable(self): """ Determine whether the file supports read() and read1() operations. Parameters ---------- N/A Returns ------- True """ return True def read(self, size=-1): """ Read bytes from the file. Parameters ---------- size : int Number of bytes to read, or -1 to read as many bytes as possible. Returns ------- data : bytes Bytes retrieved from the file. When the end of the file is reached, the length will be less than the requested size. """ start = self._pos if size in (-1, None): end = None elif size >= 0: end = start + size else: raise ValueError("invalid size: %r" % (size,)) result = b"".join(self._read_range(start, end)) self._pos += len(result) return result def read1(self, size=-1): """ Read bytes from the file. Parameters ---------- size : int Maximum number of bytes to read, or -1 to read as many bytes as possible. Returns ------- data : bytes Bytes retrieved from the file. When the end of the file is reached, the length will be zero. """ return self.read(size) def readinto(self, b): """ Read bytes from the file. Parameters ---------- b : writable bytes-like object Buffer in which to store the retrieved bytes. Returns ------- count : int Number of bytes retrieved from the file and stored in b. When the end of the file is reached, the count will be less than the requested size. """ b = memoryview(b).cast("B") start = self._pos end = start + len(b) count = 0 for chunk in self._read_range(start, end): b[count : count + len(chunk)] = chunk count += len(chunk) self._pos += count return count def readinto1(self, b): """ Read bytes from the file. Parameters ---------- b : writable bytes-like object Buffer in which to store the retrieved bytes. Returns ------- count : int Number of bytes retrieved from the file and stored in b. When the end of the file is reached, the count will be zero. """ return self.readinto(b) def seekable(self): """ Determine whether the file supports seek() and tell() operations. Parameters ---------- N/A Returns ------- True """ return True def seek(self, offset, whence=os.SEEK_SET): """ Set the current file position. Parameters ---------- offset : int Byte offset of the new file position, relative to the base position specified by whence. whence : int, optional SEEK_SET (0, default) if offset is relative to the start of the file; SEEK_CUR (1) if offset is relative to the current file position; SEEK_END (2) if offset is relative to the end of the file. Returns ------- offset : int Byte offset of the new file position. """ if whence == os.SEEK_SET: pos = offset elif whence == os.SEEK_CUR: pos = self._pos + offset elif whence == os.SEEK_END: size = self._get_size() if size is None: raise NetFileError( "size of remote file is unknown", url=self._current_url ) pos = size + offset else: raise ValueError("invalid whence: %r" % (whence,)) if pos < 0: raise ValueError("pos < 0") self._pos = pos return pos def tell(self): """ Retrieve the current file position. Parameters ---------- N/A Returns ------- offset : int Byte offset of the current file position. """ return self._pos def openurl( url, mode="r", *, buffering=-1, encoding=None, errors=None, newline=None, check_access=False, ): """ Open a URL as a random-access file object. Parameters ---------- url : str URL of the remote file. mode : str, optional Whether to access the file in text mode ('r' or 'rt'; default), or binary mode ('rb'). buffering : int, optional Buffering policy. If buffering = 0, internal buffering is disabled; each operation on the stream requires a separate request to the server. If buffering = -2, the entire file is downloaded in a single request. If buffering > 0, it specifies the minimum size of the internal buffer. If buffering = -1, the default buffer size is used. encoding : str, optional Name of character encoding used in text mode. errors : str, optional Error handling strategy used for invalid byte sequences in text mode. See the documentation of the standard "open" function for details. newline : str, optional Newline translation mode used in text mode. See the documentation of the standard "open" function for details. check_access : bool, optional If true, raise an exception immediately if the file does not exist or is not accessible. If false (default), no exception is raised until the first time you call read() or a related function. Returns ------- nf : io.IOBase A file object, implementing either the binary file API (io.BufferedIOBase) or text file API (io.TextIOBase). """ (scheme, netloc, path, _, _, _) = urllib.parse.urlparse(url) if scheme == "": raise NetFileError("no scheme specified for URL: %r" % (url,), url=url) if scheme == "file": if netloc.lower() not in ("", "localhost"): raise NetFileError("invalid file URL: %r" % (url,)) local_path = urllib.request.url2pathname(path) return open( local_path, mode, buffering=buffering, encoding=encoding, errors=errors, newline=newline, ) nf = NetFile(url, buffering=buffering) if check_access: nf._get_size() if mode == "rb": return nf elif mode == "r" or mode == "rt": return io.TextIOWrapper( nf, encoding=encoding, errors=errors, newline=newline ) else: return ValueError("invalid mode: %r" % (mode,))