Skip to content

rslearn.data_sources.planet

planet

Data source for Planet Labs API.

Planet

Bases: DataSource

A data source for Planet Labs API.

The API key should be set via environment variable (PL_API_KEY).

Source code in rslearn/data_sources/planet.py 
 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
234
235
236
237
238
239
240
241
242
243
244
245
246
class Planet(DataSource):
    """A data source for Planet Labs API.

    The API key should be set via environment variable (PL_API_KEY).
    """

    def __init__(
        self,
        item_type_id: str,
        cache_dir: str | None = None,
        asset_type_id: str = "ortho_analytic_sr",
        range_filters: dict[str, dict[str, Any]] = {},
        use_permission_filter: bool = True,
        sort_by: str | None = None,
        bands: list[str] = ["b01", "b02", "b03", "b04"],
        context: DataSourceContext = DataSourceContext(),
    ):
        """Initialize a new Planet instance.

        Args:
            item_type_id: the item type ID, like "PSScene" or "SkySatCollect".
            cache_dir: where to store downloaded assets, or None to just store it in
                temporary directory before putting into tile store.
            asset_type_id: the asset type ID, see e.g.
                https://developers.planet.com/docs/data/skysatcollect/
                for a list of SkySatCollect assets.
            range_filters: specifications for range filters to apply, such as
                {"cloud_cover": {"lte": 0.5}} to search for scenes with less than 50%
                cloud cover. It is map from the property name to a kwargs dict to apply
                when creating the range filter object.
            use_permission_filter: when querying the Planet Data API, use permission
                filter to only return scenes that we have access to.
            sort_by: name of attribute returned by Planet API to sort by like
                "-clear_percent" or "cloud_cover" (if it starts with minus sign then we
                sort descending.)
            bands: what to call the bands in the asset.
            context: the data source context.
        """
        self.item_type_id = item_type_id
        self.asset_type_id = asset_type_id
        self.range_filters = range_filters
        self.use_permission_filter = use_permission_filter
        self.sort_by = sort_by
        self.bands = bands

        if cache_dir is None:
            self.cache_dir = None
        else:
            if context.ds_path is not None:
                self.cache_dir = join_upath(context.ds_path, cache_dir)
            else:
                self.cache_dir = UPath(cache_dir)

    async def _search_items(self, geometry: STGeometry) -> list[dict[str, Any]]:
        wgs84_geometry = geometry.to_wgs84()
        geojson_data = json.loads(shapely.to_geojson(wgs84_geometry.shp))

        async with planet.Session() as session:
            client = session.client("data")
            gte = geometry.time_range[0] if geometry.time_range is not None else None
            lte = geometry.time_range[1] if geometry.time_range is not None else None
            filter_list = [
                planet.data_filter.date_range_filter("acquired", gte=gte, lte=lte),
                planet.data_filter.geometry_filter(geojson_data),
                planet.data_filter.asset_filter([self.asset_type_id]),
            ]
            if self.use_permission_filter:
                filter_list.append(planet.data_filter.permission_filter())
            for name, kwargs in self.range_filters.items():
                range_filter = planet.data_filter.range_filter(name, **kwargs)
                filter_list.append(range_filter)
            combined_filter = planet.data_filter.and_filter(filter_list)

            return [
                item
                async for item in client.search(
                    [self.item_type_id], search_filter=combined_filter
                )
            ]

    def _wrap_planet_item(self, planet_item: dict[str, Any]) -> Item:
        """Convert a decoded Planet API item into an Item object."""
        shp = shapely.geometry.shape(planet_item["geometry"])
        ts = datetime.fromisoformat(planet_item["properties"]["acquired"])
        item_geom = STGeometry(WGS84_PROJECTION, shp, (ts, ts))
        return Item(planet_item["id"], item_geom)

    def get_items(
        self, geometries: list[STGeometry], query_config: QueryConfig
    ) -> list[list[MatchedItemGroup[Item]]]:
        """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.
        """
        groups = []
        for geometry in geometries:
            planet_items = asyncio.run(self._search_items(geometry))

            if self.sort_by:
                if self.sort_by.startswith("-"):
                    multiplier = -1
                    sort_by = self.sort_by[1:]
                else:
                    multiplier = 1
                    sort_by = self.sort_by

                planet_items.sort(
                    key=lambda planet_item: multiplier
                    * planet_item["properties"][sort_by]
                )

            items = []
            for planet_item in planet_items:
                items.append(self._wrap_planet_item(planet_item))

            cur_groups = match_candidate_items_to_window(geometry, items, query_config)
            groups.append(cur_groups)

        return groups

    async def _get_item_by_name(self, name: str) -> dict[str, Any]:
        async with planet.Session() as session:
            client = session.client("data")
            filter = planet.data_filter.string_in_filter("id", [name])
            results = [
                item
                async for item in client.search(
                    [self.item_type_id], search_filter=filter
                )
            ]
            assert len(results) == 1
            return results[0]

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

        Args:
            name: the item name.

        Returns:
            the item
        """
        planet_item = asyncio.run(self._get_item_by_name(name))
        return self._wrap_planet_item(planet_item)

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

    async def _download_asset(self, item: Item, tmp_dir: pathlib.Path) -> UPath:
        """Activate asset and download it.

        Args:
            item: the item to download.
            tmp_dir: temporary directory to download the asset to in case cache_dir is
                either not set or remote.

        Returns:
            the path where the asset is downloaded.
        """
        async with planet.Session() as session:
            client = session.client("data")
            assets = await client.list_item_assets(self.item_type_id, item.name)
            asset = assets[self.asset_type_id]
            await client.activate_asset(asset)
            # Wait up to two hours for asset to be ready.
            await client.wait_asset(asset, max_attempts=1600, delay=5)

            # Need to refresh the asset so it has location attribute.
            asset = await client.get_asset(
                self.item_type_id, item.name, self.asset_type_id
            )

            if self.cache_dir is None:
                output_path = await client.download_asset(asset, directory=tmp_dir)
                return UPath(output_path)

            elif isinstance(self.cache_dir.fs, LocalFileSystem):
                output_path = await client.download_asset(
                    asset, directory=self.cache_dir.path
                )
                return UPath(output_path)

            else:
                output_path = await client.download_asset(asset, directory=tmp_dir)
                wanted_path = self.cache_dir / output_path.name
                with open(output_path, "rb") as src:
                    with wanted_path.open("wb") as dst:
                        shutil.copyfileobj(src, dst)
                return wanted_path

    def ingest(
        self,
        tile_store: TileStoreWithLayer,
        items: list[Item],
        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:
            if tile_store.is_raster_ready(item, self.bands):
                continue

            with tempfile.TemporaryDirectory() as tmp_dir:
                asset_path = asyncio.run(self._download_asset(item, Path(tmp_dir)))
                tile_store.write_raster_file(
                    item,
                    self.bands,
                    asset_path,
                    time_range=item.geometry.time_range,
                )

get_items 

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

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[Item]]]

List of groups of items that should be retrieved for each geometry.
Source code in rslearn/data_sources/planet.py 
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
def get_items(
    self, geometries: list[STGeometry], query_config: QueryConfig
) -> list[list[MatchedItemGroup[Item]]]:
    """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.
    """
    groups = []
    for geometry in geometries:
        planet_items = asyncio.run(self._search_items(geometry))

        if self.sort_by:
            if self.sort_by.startswith("-"):
                multiplier = -1
                sort_by = self.sort_by[1:]
            else:
                multiplier = 1
                sort_by = self.sort_by

            planet_items.sort(
                key=lambda planet_item: multiplier
                * planet_item["properties"][sort_by]
            )

        items = []
        for planet_item in planet_items:
            items.append(self._wrap_planet_item(planet_item))

        cur_groups = match_candidate_items_to_window(geometry, items, query_config)
        groups.append(cur_groups)

    return groups

get_item_by_name 

get_item_by_name(name: str) -> Item

Gets an item by name.

Parameters:

Name Type Description Default
name str

the item name.

 required

Returns:

Type Description
Item

the item
Source code in rslearn/data_sources/planet.py 
164
165
166
167
168
169
170
171
172
173
174
def get_item_by_name(self, name: str) -> Item:
    """Gets an item by name.

    Args:
        name: the item name.

    Returns:
        the item
    """
    planet_item = asyncio.run(self._get_item_by_name(name))
    return self._wrap_planet_item(planet_item)

deserialize_item 

deserialize_item(serialized_item: dict) -> Item

Deserializes an item from JSON-decoded data.

Source code in rslearn/data_sources/planet.py 
176
177
178
def deserialize_item(self, serialized_item: dict) -> Item:
    """Deserializes an item from JSON-decoded data."""
    return Item.deserialize(serialized_item)

ingest 

ingest(tile_store: TileStoreWithLayer, items: list[Item], 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[Item]

the items to ingest

 required
geometries list[list[STGeometry]]

a list of geometries needed for each item

 required
Source code in rslearn/data_sources/planet.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def ingest(
    self,
    tile_store: TileStoreWithLayer,
    items: list[Item],
    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:
        if tile_store.is_raster_ready(item, self.bands):
            continue

        with tempfile.TemporaryDirectory() as tmp_dir:
            asset_path = asyncio.run(self._download_asset(item, Path(tmp_dir)))
            tile_store.write_raster_file(
                item,
                self.bands,
                asset_path,
                time_range=item.geometry.time_range,
            )