Source code for sentinelhub.api.byoc

"""
Module implementing an interface with
`Sentinel Hub Bring Your Own COG API <https://docs.sentinel-hub.com/api/latest/api/byoc/>`__.
"""

# ruff: noqa: FA100
# do not use `from __future__ import annotations`, it clashes with `dataclass_json`
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, Optional, Union

from dataclasses_json import CatchAll, LetterCase, Undefined, dataclass_json
from dataclasses_json import config as dataclass_config

from ..constants import MimeType, RequestType
from ..data_collections import DataCollection
from ..geometry import Geometry
from ..types import Json, JsonDict
from .base import BaseCollection, SentinelHubFeatureIterator, SentinelHubService
from .utils import datetime_config, geometry_config, remove_undefined

CollectionType = Union["ByocCollection", DataCollection, dict, str]
TileType = Union["ByocTile", dict, str]


[docs]@dataclass_json(letter_case=LetterCase.CAMEL, undefined=Undefined.INCLUDE) @dataclass class ByocCollectionBand: """Dataclass to hold BYOC collection band specification""" source: Optional[str] = None band_index: Optional[int] = None bit_depth: int = 8 sample_format: str = "UINT" no_data: Optional[float] = None other_data: CatchAll = field(default_factory=dict)
[docs]@dataclass_json(letter_case=LetterCase.CAMEL, undefined=Undefined.INCLUDE) @dataclass class ByocCollectionAdditionalData: """Dataclass to hold BYOC collection additional data""" bands: Optional[Dict[str, ByocCollectionBand]] = None max_meters_per_pixel: Optional[float] = None max_meters_per_pixel_override: Optional[float] = None other_data: CatchAll = field(default_factory=dict)
[docs]@dataclass_json(letter_case=LetterCase.CAMEL) @dataclass class ByocCollection(BaseCollection): """Dataclass to hold BYOC collection data""" additional_data: Optional[ByocCollectionAdditionalData] = None
[docs]@dataclass_json(letter_case=LetterCase.CAMEL, undefined=Undefined.INCLUDE) @dataclass class ByocTile: """Dataclass to hold BYOC tile data""" path: str status: Optional[str] = None tile_id: Optional[str] = field(metadata=dataclass_config(field_name="id"), default=None) tile_geometry: Optional[Geometry] = field(metadata=geometry_config, default=None) cover_geometry: Optional[Geometry] = field(metadata=geometry_config, default=None) created: Optional[datetime] = field(metadata=datetime_config, default=None) sensing_time: Optional[datetime] = field(metadata=datetime_config, default=None) ingestion_start: Optional[datetime] = field(metadata=datetime_config, default=None) additional_data: Optional[dict] = None other_data: CatchAll = field(default_factory=dict)
[docs]class SentinelHubBYOC(SentinelHubService): """An interface class for Sentinel Hub Bring your own COG (BYOC) API For more info check `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#tag/byoc_collection>`__. """ @staticmethod def _get_service_url(base_url: str) -> str: """Provides URL to Catalog API""" return f"{base_url}/api/v1/byoc"
[docs] def iter_collections(self, search: Optional[str] = None, **kwargs: Any) -> SentinelHubFeatureIterator: """Retrieve collections `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/getByocCollections>`__ :param search: A search query :param kwargs: Any other request parameters :return: iterator over collections """ return SentinelHubFeatureIterator( client=self.client, url=f"{self.service_url}/collections", params={"search": search, **kwargs}, exception_message="Failed to obtain information about available BYOC collections", )
[docs] def get_collection(self, collection: CollectionType) -> JsonDict: """Get collection by its id `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/getByocCollectionById>`__ :param collection: a ByocCollection, dict or collection id string :return: dictionary of the collection """ url = f"{self.service_url}/collections/{self._parse_id(collection)}" return self.client.get_json_dict(url=url, use_session=True, extract_key="data")
[docs] def create_collection(self, collection: CollectionType) -> JsonDict: """Create a new collection `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/createByocCollection>`__ :param collection: ByocCollection object or a dictionary :return: dictionary of the created collection """ coll = self._to_dict(collection) url = f"{self.service_url}/collections" return self.client.get_json_dict(url=url, post_values=coll, use_session=True, extract_key="data")
[docs] def update_collection(self, collection: CollectionType) -> Json: """Update an existing collection `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/updateByocCollectionById>`__ :param collection: ByocCollection object or a dictionary """ coll = self._to_dict(collection) url = f"{self.service_url}/collections/{self._parse_id(coll)}" headers = {"Content-Type": MimeType.JSON.get_string()} return self.client.get_json( url=url, request_type=RequestType.PUT, post_values=coll, headers=headers, use_session=True )
[docs] def delete_collection(self, collection: CollectionType) -> Json: """Delete existing collection by its id `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/deleteByocCollectionById>`__ :param collection: a ByocCollection, dict or collection id string """ url = f"{self.service_url}/collections/{self._parse_id(collection)}" return self.client.get_json(url=url, request_type=RequestType.DELETE, use_session=True)
[docs] def copy_tiles(self, from_collection: CollectionType, to_collection: CollectionType) -> Json: """Copy tiles from one collection to another `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/copyByocCollectionTiles>`__ :param from_collection: a ByocCollection, dict or collection id string :param to_collection: a ByocCollection, dict or collection id string """ url = ( f"{self.service_url}/collections/{self._parse_id(from_collection)}" f"/copyTiles?toCollection={self._parse_id(to_collection)}" ) return self.client.get_json(url=url, request_type=RequestType.POST, use_session=True)
[docs] def iter_tiles( self, collection: CollectionType, sort: Optional[str] = None, path: Optional[str] = None, **kwargs: Any ) -> SentinelHubFeatureIterator: """Iterator over collection tiles `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/getByocCollectionTiles>`__ :param collection: a ByocCollection, dict or collection id string :param sort: Order in which to return tiles :param path: An exact path where tiles are located :param kwargs: Any other request parameters :return: An iterator over payloads of tiles from the collection """ collection_id = self._parse_id(collection) return SentinelHubFeatureIterator( client=self.client, url=f"{self.service_url}/collections/{collection_id}/tiles", params={"sort": sort, "path": path, **kwargs}, exception_message=f"Failed to obtain information about tiles in BYOC collection {collection_id}", )
[docs] def get_tile(self, collection: CollectionType, tile: TileType) -> JsonDict: """Get a tile of collection `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/getByocCollectionTileById>`__ :param collection: a ByocCollection, dict or collection id string :param tile: a ByocTile, dict or tile id string :return: dictionary of the tile """ url = f"{self.service_url}/collections/{self._parse_id(collection)}/tiles/{self._parse_id(tile)}" return self.client.get_json_dict(url=url, use_session=True, extract_key="data")
[docs] def create_tile(self, collection: CollectionType, tile: TileType) -> JsonDict: """Create tile within collection `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/createByocCollectionTile>`__ :param collection: a ByocCollection, dict or collection id string :param tile: a ByocTile or dict :return: dictionary of the tile """ _tile = self._to_dict(tile) url = f"{self.service_url}/collections/{self._parse_id(collection)}/tiles" return self.client.get_json_dict(url=url, post_values=_tile, use_session=True, extract_key="data")
[docs] def update_tile(self, collection: CollectionType, tile: TileType) -> Json: """Update a tile within collection `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/updateByocCollectionTileById>`__ :param collection: a ByocCollection, dict or collection id string :param tile: a ByocTile or dict """ url = f"{self.service_url}/collections/{self._parse_id(collection)}/tiles/{self._parse_id(tile)}" headers = {"Content-Type": MimeType.JSON.get_string()} _tile = self._to_dict(tile) updates = remove_undefined( { "path": _tile["path"], "coverGeometry": _tile.get("coverGeometry"), "sensingTime": _tile.get("sensingTime"), } ) return self.client.get_json( url=url, request_type=RequestType.PUT, post_values=updates, headers=headers, use_session=True )
[docs] def delete_tile(self, collection: CollectionType, tile: TileType) -> Json: """Delete a tile from collection `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/deleteByocCollectionTileById>`__ :param collection: a ByocCollection, dict or collection id string :param tile: a ByocTile, dict or tile id string """ url = f"{self.service_url}/collections/{self._parse_id(collection)}/tiles/{self._parse_id(tile)}" return self.client.get_json(url=url, request_type=RequestType.DELETE, use_session=True)
[docs] def reingest_tile(self, collection: CollectionType, tile: TileType) -> Json: """Re-ingests a tile into a collection `BYOC API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/reingestByocCollectionTileById>`__ :param collection: a ByocCollection, dict or collection id string :param tile: a ByocTile, dict or tile id string """ url = f"{self.service_url}/collections/{self._parse_id(collection)}/tiles/{self._parse_id(tile)}/reingest" return self.client.get_json(url=url, request_type=RequestType.POST, use_session=True)
@staticmethod def _parse_id(data: object) -> Optional[str]: if isinstance(data, (ByocCollection, DataCollection)): return data.collection_id if isinstance(data, ByocTile): return data.tile_id if isinstance(data, dict): return data["id"] if isinstance(data, str): return data raise ValueError(f"Expected a BYOC/Data dataclass, dictionary or a string, got {data}.") @staticmethod def _to_dict(data: object) -> dict: """Constructs dict from an object (either dataclass or dict)""" if isinstance(data, (ByocCollection, ByocTile, ByocCollectionAdditionalData, ByocCollectionBand)): return data.to_dict() # type: ignore[union-attr] # to_dict method comes from decorators and is undetectable if isinstance(data, dict): return data raise ValueError(f"Expected either a data class (e.g., ByocCollection and similar) or a dict, got {data}.")