Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from __future__ import print_function
from gbdxtools.images.rda_image import RDAImage
from gbdxtools.rda.interface import RDA
from gbdxtools.rda.util import ortho_params
rda = RDA()
class S3Image(RDAImage):
'''Dask based access to geotiffs on S3.
Args:
path (string): path to the geotiff file in S3.
proj (string): destination EPSG string, e.g. 'EPSG:4326'
Perform optional reprojection if needed.
src_proj (string): source EPSG string
Define the source projection if it can't be automatically determined.
Example:
>>> img = S3Image('landsat-pds/c1/L8/139/045/LC08_L1TP_139045_20170304_20170316_01_T1/LC08_L1TP_139045_20170304_20170316_01_T1_B3.TIF')'''
def __new__(cls, path, **kwargs):
path = '/vsis3/{}'.format(path)
graph = cls._build_graph(path, kwargs.get("proj", None), kwargs.get("src_proj", None))
try:
def __call__(self, *args, **kwargs):
if len(args) > 0 and all([isinstance(arg, gbdx.images.rda_image.RDAImage) for arg in args]):
return self._rda_image_call(*args, **kwargs)
self._nodes = [ContentHashedDict({
"operator": self._operator,
"_ancestors": [arg._id for arg in args],
"parameters": OrderedDict({
k:json.dumps(v, sort_keys=True) if not isinstance(v, basestring) else v
for k,v in sorted(kwargs.items(), key=lambda x: x[0])})
})]
for arg in args:
self._nodes.extend(arg._nodes)
self._edges = [ContentHashedDict({"index": idx + 1, "source": arg._nodes[0]._id,
"destination": self._nodes[0]._id})
for idx, arg in enumerate(args)]
for arg in args:
self._edges.extend(arg._edges)
def _rda_image_call(self, *args, **kwargs):
out = self(*[arg.rda for arg in args], **kwargs)
rda_img = gbdx.images.rda_image.RDAImage(out)
return rda_img
from __future__ import print_function
from gbdxtools.images.rda_image import RDAImage
class RDABaseImage(RDAImage):
__rda_id__ = None
def __new__(cls, rda_id=None, **kwargs):
cls = cls.__Driver__(rda_id=rda_id, **kwargs).drive(cls, **kwargs)
self = super(RDABaseImage, cls).__new__(cls, cls.__driver__.payload, **kwargs)
return self.__post_new_hook__(**kwargs)
def __post_new_hook__(self, **kwargs):
return self.aoi(**kwargs)
@property
def options(self):
return self.__driver__.options
@property
def _graph(self):
return self.__driver__.graph
def __getitem__(self, geometry):
im = super(RDAImage, self).__getitem__(geometry)
if isinstance(im, GeoDaskImage):
im._rda_op = self._rda_op
return im
def __new__(cls, name, node_id=None, **kwargs):
return RDAImage(TemplateMeta(name, node_id, **kwargs))