Source code for rastervision.core.data.raster_source.rasterio_source_config
from typing import List, Union
from rastervision.core.box import Box
from rastervision.core.data.raster_source import RasterSourceConfig, RasterioSource
from rastervision.pipeline.config import ConfigError, Field, register_config
def rasterio_source_config_upgrader(cfg_dict: dict, version: int) -> dict:
if version == 5:
# removed in version 6
x_shift = cfg_dict.get('x_shift', 0)
y_shift = cfg_dict.get('y_shift', 0)
if x_shift != 0 or y_shift != 0:
raise ConfigError('x_shift and y_shift are deprecated. '
'Use the ShiftTransformer instead.')
try:
del cfg_dict['x_shift']
del cfg_dict['y_shift']
except KeyError:
pass
return cfg_dict
[docs]@register_config('rasterio_source', upgrader=rasterio_source_config_upgrader)
class RasterioSourceConfig(RasterSourceConfig):
"""Configure a :class:`.RasterioSource`."""
uris: Union[str, List[str]] = Field(
...,
description='One or more image URIs that comprise the imagery for a '
'scene. The format of each file can be any that can be read by '
'Rasterio/GDAL. If > 1 URI is provided, a VRT will be created to '
'mosaic together the individual images.')
allow_streaming: bool = Field(
False,
description='Stream assets as needed rather than downloading them.')
[docs] def build(self, tmp_dir, use_transformers=True):
raster_transformers = ([rt.build() for rt in self.transformers]
if use_transformers else [])
bbox = Box(*self.bbox) if self.bbox is not None else None
return RasterioSource(
uris=self.uris,
raster_transformers=raster_transformers,
tmp_dir=tmp_dir,
allow_streaming=self.allow_streaming,
channel_order=self.channel_order,
bbox=bbox)