1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
|
"""FilePath tests. MemoryFile requires GDAL 2.0+.
Tests in this file will ONLY run for GDAL >= 3.x"""
# TODO: delete at version 2.0. FilePath is deprecated in version 1.4.
from io import BytesIO
import logging
import os.path
import pytest
import rasterio
from rasterio.enums import MaskFlags
from rasterio.shutil import copyfiles
from rasterio.windows import Window
try:
from rasterio.io import FilePath
except ImportError:
pytest.skip("FilePath is not available for GDAL <3.0", allow_module_level=True)
@pytest.fixture(scope='function')
def rgb_lzw_file_object(path_rgb_lzw_byte_tif):
"""Get the open file of our RGB.bytes.tif file."""
return open(path_rgb_lzw_byte_tif, 'rb')
@pytest.fixture(scope='function')
def rgb_file_object(path_rgb_byte_tif):
"""Get RGB.bytes.tif file opened in 'rb' mode"""
return open(path_rgb_byte_tif, 'rb')
def test_initial_empty():
with pytest.raises(TypeError):
FilePath()
def test_initial_not_file_str():
"""Creating from not file-like fails."""
with pytest.raises(TypeError):
FilePath("lolwut")
def test_initial_not_file_bytes():
"""Creating from not file-like fails."""
with pytest.raises(TypeError):
FilePath(b'lolwut')
def test_initial_bytes(rgb_file_object):
"""FilePath contents can initialized from bytes and opened."""
with FilePath(rgb_file_object) as vsifile:
with vsifile.open() as src:
assert src.driver == 'GTiff'
assert src.count == 3
assert src.dtypes == ("uint8", "uint8", "uint8")
assert src.read().shape == (3, 718, 791)
def test_initial_bytes_boundless(rgb_file_object):
"""FilePath contents can initialized from bytes and opened."""
with FilePath(rgb_file_object) as vsifile:
with vsifile.open() as src:
assert src.driver == "GTiff"
assert src.count == 3
assert src.dtypes == ("uint8", "uint8", "uint8")
assert src.read(window=Window(0, 0, 800, 800), boundless=True).shape == (
3,
800,
800,
)
def test_filepath_vrt(rgb_file_object):
"""A FilePath can be wrapped by a VRT."""
from rasterio.vrt import _boundless_vrt_doc
with FilePath(rgb_file_object) as vsifile, vsifile.open() as dst:
vrt_doc = _boundless_vrt_doc(dst)
with rasterio.open(vrt_doc) as src:
assert src.driver == "VRT"
assert src.count == 3
assert src.dtypes == ('uint8', 'uint8', 'uint8')
assert src.read().shape == (3, 718, 791)
def test_initial_lzw_bytes(rgb_lzw_file_object):
"""FilePath contents can initialized from bytes and opened."""
with FilePath(rgb_lzw_file_object) as vsifile:
with vsifile.open() as src:
assert src.driver == 'GTiff'
assert src.count == 3
assert src.dtypes == ('uint8', 'uint8', 'uint8')
assert src.read().shape == (3, 718, 791)
def test_initial_file_object(rgb_file_object):
"""FilePath contents can initialized from bytes and opened."""
with FilePath(rgb_file_object) as vsifile:
with vsifile.open() as src:
assert src.driver == 'GTiff'
assert src.count == 3
assert src.dtypes == ('uint8', 'uint8', 'uint8')
assert src.read().shape == (3, 718, 791)
def test_closed(rgb_file_object):
"""A closed FilePath can not be opened."""
with FilePath(rgb_file_object) as vsifile:
pass
with pytest.raises(IOError):
vsifile.open()
def test_file_object_read(rgb_file_object):
"""An example of reading from a file object"""
with rasterio.open(rgb_file_object) as src:
assert src.driver == 'GTiff'
assert src.count == 3
assert src.dtypes == ('uint8', 'uint8', 'uint8')
assert src.read().shape == (3, 718, 791)
def test_file_object_read_variant(rgb_file_object):
"""An example of reading from a FilePath object"""
with rasterio.open(FilePath(rgb_file_object)) as src:
assert src.driver == 'GTiff'
assert src.count == 3
assert src.dtypes == ('uint8', 'uint8', 'uint8')
assert src.read().shape == (3, 718, 791)
def test_file_object_read_variant2(rgb_file_object):
"""An example of reading from a BytesIO object version of a file's contents."""
with rasterio.open(BytesIO(rgb_file_object.read())) as src:
assert src.driver == 'GTiff'
assert src.count == 3
assert src.dtypes == ('uint8', 'uint8', 'uint8')
assert src.read().shape == (3, 718, 791)
def test_vrt_vsifile(data_dir, path_white_gemini_iv_vrt):
"""Successfully read an in-memory VRT"""
with open(path_white_gemini_iv_vrt) as vrtfile:
source = vrtfile.read()
source = source.replace(
'<SourceFilename relativeToVRT="1">389225main_sw_1965_1024.jpg</SourceFilename>',
f'<SourceFilename relativeToVRT="0">{data_dir}/389225main_sw_1965_1024.jpg</SourceFilename>',
)
source = BytesIO(source.encode("utf-8"))
with FilePath(source) as vsifile:
with vsifile.open() as src:
assert src.driver == 'VRT'
assert src.count == 3
assert src.dtypes == ('uint8', 'uint8', 'uint8')
assert src.read().shape == (3, 768, 1024)
@pytest.mark.xfail(reason="Copying is not supported by FilePath")
def test_vsifile_copyfiles(path_rgb_msk_byte_tif):
"""Multiple files can be copied to a FilePath using copyfiles"""
with rasterio.open(path_rgb_msk_byte_tif) as src:
src_basename = os.path.basename(src.name)
with FilePath(dirname="foo", filename=src_basename) as vsifile:
copyfiles(src.name, vsifile.name)
with vsifile.open() as rgb2:
assert sorted(rgb2.files) == sorted(
[f"/vsimem/foo/{src_basename}", f"/vsimem/foo/{src_basename}.msk"]
)
@pytest.mark.xfail(reason="FilePath does not implement '.files' property properly.")
def test_multi_vsifile(path_rgb_msk_byte_tif):
"""Multiple files can be copied to a FilePath using copyfiles"""
with open(path_rgb_msk_byte_tif, "rb") as tif_fp, open(
path_rgb_msk_byte_tif + ".msk", "rb"
) as msk_fp:
with FilePath(
tif_fp, dirname="bar", filename="foo.tif"
) as tifvsifile, FilePath(msk_fp, dirname="bar", filename="foo.tif.msk"):
with tifvsifile.open() as src:
assert sorted(os.path.basename(fn) for fn in src.files) == sorted(['foo.tif', 'foo.tif.msk'])
assert src.mask_flag_enums == ([MaskFlags.per_dataset],) * 3
def _open_geotiff(file_path):
with open(file_path, 'rb') as file_obj:
with rasterio.open(file_obj) as dataset:
dataset.read()
def test_concurrent(path_rgb_byte_tif, path_rgb_lzw_byte_tif, path_cogeo_tif, path_alpha_tif):
"""Test multiple threads opening multiple files at the same time."""
from concurrent.futures import ThreadPoolExecutor
tifs = [path_rgb_byte_tif, path_rgb_lzw_byte_tif, path_cogeo_tif, path_alpha_tif] * 4
with ThreadPoolExecutor(max_workers=8) as exe:
list(exe.map(_open_geotiff, tifs, timeout=5))
def test_python_file_reuse():
"""Test that we can reuse a Python file, see gh-2550."""
ascii_raster_string = """ncols 5
nrows 5
xllcorner 440720.000000000000
yllcorner 3750120.000000000000
cellsize 60.000000000000
nodata_value -99999
107 123 132 115 132
115 132 107 123 148
115 132 140 132 123
148 132 123 123 115
132 156 132 140 132
"""
ascii_raster_io = BytesIO(ascii_raster_string.encode("utf-8"))
with rasterio.open(ascii_raster_io) as rds:
_ = rds.bounds
with rasterio.open(ascii_raster_io) as rds:
_ = rds.bounds
def test_quieter_vsi_plugin_notifications(caplog, path_rgb_byte_tif):
"""Expect no warning or error level log messages about .aux or .hdr files."""
with caplog.at_level(logging.WARNING):
with open(path_rgb_byte_tif, "rb") as f, FilePath(f) as vsi_file:
with vsi_file.open() as src:
_ = src.profile
assert "not found in virtual filesystem" not in caplog.text
|
