Last active
May 22, 2018 23:46
-
-
Save pedros007/55c6e33224596fb4d8e9e6b68b24ed9b to your computer and use it in GitHub Desktop.
gdal /vsimem/ file writer. This is compatible with boto s3.Object#put(Body=SimpleVSIMemFile('/vsimem/foo.tif')
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class SimpleVSIMemFileError(Exception): | |
"""Unknown SimpleVSIMemFile error with VSI subsystem.""" | |
class SimpleVSIMEMFile(object): | |
def __init__(self, path): | |
"""Simple file-like object for reading out of a VSIMEM dataset. | |
Params: | |
path: /vsimem path to use | |
""" | |
self._path = path | |
self._size = gdal.VSIStatL(self._path).size | |
self._check_error() | |
self._pos = 0 | |
def __len__(self): | |
"""Length of the file.""" | |
return self._size | |
def read(self, size=-1): | |
"""Read size bytes from the file. | |
Params: | |
size: Number of bytes to read. | |
""" | |
length = len(self) | |
if self._pos >= length: | |
# No more data to read | |
return b"" | |
if size == -1: | |
# Set size to remainder of file | |
size = length - self._pos | |
else: | |
# Limit size to remainder of file | |
size = min(size, length - self._pos) | |
# Open file | |
vsif = gdal.VSIFOpenL(self._path, "r") | |
self._check_error() | |
try: | |
# Seek to current position, read data, and update position | |
gdal.VSIFSeekL(vsif, self._pos, 0) | |
self._check_error() | |
buf = gdal.VSIFReadL(1, size, vsif) | |
self._check_error() | |
self._pos += len(buf) | |
finally: | |
# Close file | |
gdal.VSIFCloseL(vsif) | |
self._check_error() | |
return buf | |
def seek(self, offset, whence=0): | |
"""Seek to position in file.""" | |
if whence == 0: | |
# Seek from start of file | |
self._pos = min(offset, len(self)) | |
elif whence == 1: | |
# Seek from current position | |
self._pos = min(max(0, self._pos + offset), len(self)) | |
elif whence == 2: | |
# Seek from end of file | |
self._pos = max(0, len(self) - offset) | |
return self._pos | |
def tell(self): | |
"""Tell current position in file.""" | |
return self._pos | |
def _check_error(self): | |
if gdal.VSIGetLastErrorNo() != 0: | |
raise SimpleVSIMemFileError(gdal.VSIGetLastErrorMsg()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment