Skip to content

rslearn.data_sources.aws_sentinel1

aws_sentinel1

Data source for Sentinel-1 on AWS.

Sentinel1

Bases: DataSource, TileStore

A data source for Sentinel-1 GRD imagery on AWS.

Specifically, uses the sentinel-s1-l1c S3 bucket maintained by Sinergise. See https://aws.amazon.com/marketplace/pp/prodview-uxrsbvhd35ifw for details about the bucket.

We use the Copernicus API for metadata search. So the bucket is only used for downloading the images.

Currently, it only supports GRD IW DV scenes.

Source code in rslearn/data_sources/aws_sentinel1.py 
 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
class Sentinel1(DataSource, TileStore):
    """A data source for Sentinel-1 GRD imagery on AWS.

    Specifically, uses the sentinel-s1-l1c S3 bucket maintained by Sinergise. See
    https://aws.amazon.com/marketplace/pp/prodview-uxrsbvhd35ifw for details about the
    bucket.

    We use the Copernicus API for metadata search. So the bucket is only used for
    downloading the images.

    Currently, it only supports GRD IW DV scenes.
    """

    bucket_name = "sentinel-s1-l1c"
    bands = ["vv", "vh"]

    def __init__(
        self,
        orbit_direction: Sentinel1OrbitDirection | None = None,
        context: DataSourceContext = DataSourceContext(),
    ) -> None:
        """Initialize a new Sentinel1 instance.

        Args:
            orbit_direction: optional orbit direction to filter by.
            context: the data source context.
        """
        self.client = boto3.client("s3")
        self.bucket = boto3.resource("s3").Bucket(self.bucket_name)
        self.sentinel1 = CopernicusSentinel1(
            product_type=Sentinel1ProductType.IW_GRDH,
            polarisation=Sentinel1Polarisation.VV_VH,
            orbit_direction=orbit_direction,
        )

    def get_items(
        self, geometries: list[STGeometry], query_config: QueryConfig
    ) -> list[list[MatchedItemGroup[CopernicusItem]]]:
        """Get a list of items in the data source intersecting the given geometries.

        Args:
            geometries: the spatiotemporal geometries
            query_config: the query configuration

        Returns:
            List of groups of items that should be retrieved for each geometry.
        """
        return self.sentinel1.get_items(geometries, query_config)

    def get_item_by_name(self, name: str) -> CopernicusItem:
        """Gets an item by name."""
        return self.sentinel1.get_item_by_name(name)

    def deserialize_item(self, serialized_item: dict) -> CopernicusItem:
        """Deserializes an item from JSON-decoded data."""
        return CopernicusItem.deserialize(serialized_item)

    def ingest(
        self,
        tile_store: TileStoreWithLayer,
        items: list[CopernicusItem],
        geometries: list[list[STGeometry]],
    ) -> None:
        """Ingest items into the given tile store.

        Args:
            tile_store: the tile store to ingest into
            items: the items to ingest
            geometries: a list of geometries needed for each item
        """
        for item in items:
            for band in self.bands:
                band_names = [band]
                if tile_store.is_raster_ready(item, band_names):
                    continue

                # Item name is like "S1C_IW_GRDH_1SDV_20250528T172106_20250528T172131_002534_00545C_B433.SAFE".
                item_name_prefix = item.name.split(".")[0]
                time_str = item_name_prefix.split("_")[4]
                if len(time_str) != 15:
                    raise ValueError(
                        f"expected 15-character time string but got {time_str}"
                    )
                # We convert to int here since path in bucket isn't padded with leading 0s.
                year = int(time_str[0:4])
                month = int(time_str[4:6])
                day = int(time_str[6:8])
                blob_path = f"GRD/{year}/{month}/{day}/IW/DV/{item_name_prefix}/measurement/iw-{band}.tiff"

                with tempfile.TemporaryDirectory() as tmp_dir:
                    fname = os.path.join(tmp_dir, f"{band}.tif")
                    try:
                        self.bucket.download_file(
                            blob_path,
                            fname,
                            ExtraArgs={"RequestPayer": "requester"},
                        )
                    except:
                        logger.error(
                            f"encountered error while downloading s3://{self.bucket_name}/{blob_path}"
                        )
                        raise
                    tile_store.write_raster_file(
                        item,
                        band_names,
                        UPath(fname),
                        time_range=item.geometry.time_range,
                    )

get_items 

get_items(geometries: list[STGeometry], query_config: QueryConfig) -> list[list[MatchedItemGroup[CopernicusItem]]]

Get a list of items in the data source intersecting the given geometries.

Parameters:

Name Type Description Default
geometries list[STGeometry]

the spatiotemporal geometries

 required
query_config QueryConfig

the query configuration

 required

Returns:

Type Description
list[list[MatchedItemGroup[CopernicusItem]]]

List of groups of items that should be retrieved for each geometry.
Source code in rslearn/data_sources/aws_sentinel1.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def get_items(
    self, geometries: list[STGeometry], query_config: QueryConfig
) -> list[list[MatchedItemGroup[CopernicusItem]]]:
    """Get a list of items in the data source intersecting the given geometries.

    Args:
        geometries: the spatiotemporal geometries
        query_config: the query configuration

    Returns:
        List of groups of items that should be retrieved for each geometry.
    """
    return self.sentinel1.get_items(geometries, query_config)

get_item_by_name 

get_item_by_name(name: str) -> CopernicusItem

Gets an item by name.

Source code in rslearn/data_sources/aws_sentinel1.py 
77
78
79
def get_item_by_name(self, name: str) -> CopernicusItem:
    """Gets an item by name."""
    return self.sentinel1.get_item_by_name(name)

deserialize_item 

deserialize_item(serialized_item: dict) -> CopernicusItem

Deserializes an item from JSON-decoded data.

Source code in rslearn/data_sources/aws_sentinel1.py 
81
82
83
def deserialize_item(self, serialized_item: dict) -> CopernicusItem:
    """Deserializes an item from JSON-decoded data."""
    return CopernicusItem.deserialize(serialized_item)

ingest 

ingest(tile_store: TileStoreWithLayer, items: list[CopernicusItem], geometries: list[list[STGeometry]]) -> None

Ingest items into the given tile store.

Parameters:

Name Type Description Default
tile_store TileStoreWithLayer

the tile store to ingest into

 required
items list[CopernicusItem]

the items to ingest

 required
geometries list[list[STGeometry]]

a list of geometries needed for each item

 required
Source code in rslearn/data_sources/aws_sentinel1.py 
 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
def ingest(
    self,
    tile_store: TileStoreWithLayer,
    items: list[CopernicusItem],
    geometries: list[list[STGeometry]],
) -> None:
    """Ingest items into the given tile store.

    Args:
        tile_store: the tile store to ingest into
        items: the items to ingest
        geometries: a list of geometries needed for each item
    """
    for item in items:
        for band in self.bands:
            band_names = [band]
            if tile_store.is_raster_ready(item, band_names):
                continue

            # Item name is like "S1C_IW_GRDH_1SDV_20250528T172106_20250528T172131_002534_00545C_B433.SAFE".
            item_name_prefix = item.name.split(".")[0]
            time_str = item_name_prefix.split("_")[4]
            if len(time_str) != 15:
                raise ValueError(
                    f"expected 15-character time string but got {time_str}"
                )
            # We convert to int here since path in bucket isn't padded with leading 0s.
            year = int(time_str[0:4])
            month = int(time_str[4:6])
            day = int(time_str[6:8])
            blob_path = f"GRD/{year}/{month}/{day}/IW/DV/{item_name_prefix}/measurement/iw-{band}.tiff"

            with tempfile.TemporaryDirectory() as tmp_dir:
                fname = os.path.join(tmp_dir, f"{band}.tif")
                try:
                    self.bucket.download_file(
                        blob_path,
                        fname,
                        ExtraArgs={"RequestPayer": "requester"},
                    )
                except:
                    logger.error(
                        f"encountered error while downloading s3://{self.bucket_name}/{blob_path}"
                    )
                    raise
                tile_store.write_raster_file(
                    item,
                    band_names,
                    UPath(fname),
                    time_range=item.geometry.time_range,
                )