diff --git a/pystac/__init__.py b/pystac/__init__.py index c029ac28..c2a5b53b 100644 --- a/pystac/__init__.py +++ b/pystac/__init__.py @@ -33,6 +33,7 @@ "RangeSummary", "Item", "Asset", + "ItemAssetDefinition", "ItemCollection", "Provider", "ProviderRole", @@ -81,6 +82,7 @@ from pystac.summaries import RangeSummary, Summaries from pystac.asset import Asset from pystac.item import Item +from pystac.item_assets import ItemAssetDefinition from pystac.item_collection import ItemCollection from pystac.provider import ProviderRole, Provider from pystac.utils import HREF diff --git a/pystac/collection.py b/pystac/collection.py index 9c0c24b7..5323b82c 100644 --- a/pystac/collection.py +++ b/pystac/collection.py @@ -20,6 +20,7 @@ from pystac.asset import Asset, Assets from pystac.catalog import Catalog from pystac.errors import DeprecatedWarning, ExtensionNotImplemented, STACTypeError +from pystac.item_assets import ItemAssetDefinition, _ItemAssets from pystac.layout import HrefLayoutStrategy from pystac.link import Link from pystac.provider import Provider @@ -553,6 +554,7 @@ def __init__( self.keywords = keywords self.providers = providers self.summaries = summaries or Summaries.empty() + self._item_assets: _ItemAssets | None = None self.assets = {} if assets is not None: @@ -731,6 +733,27 @@ def get_item(self, id: str, recursive: bool = False) -> Item | None: return super().get_item(id, recursive=recursive) raise e + @property + def item_assets(self) -> dict[str, ItemAssetDefinition] | None: + if self._item_assets is None and "item_assets" in self.extra_fields: + self._item_assets = _ItemAssets(self) + return self._item_assets + + @item_assets.setter + def item_assets( + self, item_assets: dict[str, ItemAssetDefinition | dict[str, Any]] | None + ) -> None: + # clear out the cached value + self._item_assets = None + + if item_assets is None: + self.extra_fields.pop("item_assets") + else: + self.extra_fields["item_assets"] = { + k: v if isinstance(v, dict) else v.to_dict() + for k, v in item_assets.items() + } + def update_extent_from_items(self) -> None: """ Update datetime and bbox based on all items to a single bbox and time window. diff --git a/pystac/extensions/base.py b/pystac/extensions/base.py index 1b3016fe..241c8381 100644 --- a/pystac/extensions/base.py +++ b/pystac/extensions/base.py @@ -5,7 +5,6 @@ from abc import ABC, abstractmethod from collections.abc import Iterable from typing import ( - TYPE_CHECKING, Any, Generic, TypeVar, @@ -14,9 +13,6 @@ import pystac -if TYPE_CHECKING: - from pystac.extensions.item_assets import AssetDefinition - VERSION_REGEX = re.compile("/v[0-9].[0-9].*/") @@ -158,7 +154,7 @@ def has_extension(cls, obj: S) -> bool: @classmethod def validate_owner_has_extension( cls, - asset: pystac.Asset | AssetDefinition, + asset: pystac.Asset | pystac.ItemAssetDefinition, add_if_missing: bool = False, ) -> None: """ @@ -190,7 +186,7 @@ def validate_owner_has_extension( @classmethod def ensure_owner_has_extension( cls, - asset_or_link: pystac.Asset | AssetDefinition | pystac.Link, + asset_or_link: pystac.Asset | pystac.ItemAssetDefinition | pystac.Link, add_if_missing: bool = False, ) -> None: """Given an :class:`~pystac.Asset`, checks if the asset's owner has this diff --git a/pystac/extensions/classification.py b/pystac/extensions/classification.py index bd8b4557..beb30bbe 100644 --- a/pystac/extensions/classification.py +++ b/pystac/extensions/classification.py @@ -16,7 +16,6 @@ ) import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -27,7 +26,7 @@ from pystac.serialization.identify import STACJSONDescription, STACVersionID from pystac.utils import get_required, map_opt -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition, RasterBand) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition, RasterBand) SCHEMA_URI_PATTERN: str = ( "https://stac-extensions.github.io/classification/v{version}/schema.json" @@ -492,7 +491,7 @@ class ClassificationExtension( """An abstract class that can be used to extend the properties of :class:`~pystac.Item`, :class:`~pystac.Asset`, :class:`~pystac.extension.raster.RasterBand`, or - :class:`~pystac.extension.item_assets.AssetDefinition` with properties from the + :class:`~pystac.ItemAssetDefinition` with properties from the :stac-ext:`Classification Extension `. This class is generic over the type of STAC object being extended. @@ -600,7 +599,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> ClassificationExtension[T] This extension can be applied to instances of :class:`~pystac.Item`, :class:`~pystac.Asset`, - :class:`~pystac.extensions.item_assets.AssetDefinition`, or + :class:`~pystac.ItemAssetDefinition`, or :class:`~pystac.extension.raster.RasterBand`. Raises: @@ -612,7 +611,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> ClassificationExtension[T] elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(ClassificationExtension[T], AssetClassificationExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast( ClassificationExtension[T], ItemAssetsClassificationExtension(obj) @@ -663,17 +662,19 @@ def __repr__(self) -> str: class ItemAssetsClassificationExtension( - ClassificationExtension[item_assets.AssetDefinition] + ClassificationExtension[pystac.ItemAssetDefinition] ): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties def __repr__(self) -> str: - return f" DatacubeExtension[T]: elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(DatacubeExtension[T], AssetDatacubeExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(DatacubeExtension[T], ItemAssetsDatacubeExtension(obj)) else: @@ -691,11 +690,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsDatacubeExtension(DatacubeExtension[item_assets.AssetDefinition]): +class ItemAssetsDatacubeExtension(DatacubeExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/eo.py b/pystac/extensions/eo.py index 54b6cbcb..c9628a3c 100644 --- a/pystac/extensions/eo.py +++ b/pystac/extensions/eo.py @@ -14,7 +14,7 @@ ) import pystac -from pystac.extensions import item_assets, projection, view +from pystac.extensions import projection, view from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -25,7 +25,7 @@ from pystac.summaries import RangeSummary from pystac.utils import get_required, map_opt -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI: str = "https://stac-extensions.github.io/eo/v1.1.0/schema.json" SCHEMA_URIS: list[str] = [ @@ -409,7 +409,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> EOExtension[T]: elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(EOExtension[T], AssetEOExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(EOExtension[T], ItemAssetsEOExtension(obj)) else: @@ -536,9 +536,9 @@ def __repr__(self) -> str: return f"" -class ItemAssetsEOExtension(EOExtension[item_assets.AssetDefinition]): +class ItemAssetsEOExtension(EOExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition def _get_bands(self) -> list[Band] | None: if BANDS_PROP not in self.properties: @@ -550,7 +550,7 @@ def _get_bands(self) -> list[Band] | None: ) ) - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/ext.py b/pystac/extensions/ext.py index 2488b1c4..581886c1 100644 --- a/pystac/extensions/ext.py +++ b/pystac/extensions/ext.py @@ -1,13 +1,21 @@ from dataclasses import dataclass from typing import Any, Generic, Literal, TypeVar, cast -from pystac import Asset, Catalog, Collection, Item, Link, STACError +from pystac import ( + Asset, + Catalog, + Collection, + Item, + ItemAssetDefinition, + Link, + STACError, +) from pystac.extensions.classification import ClassificationExtension from pystac.extensions.datacube import DatacubeExtension from pystac.extensions.eo import EOExtension from pystac.extensions.file import FileExtension from pystac.extensions.grid import GridExtension -from pystac.extensions.item_assets import AssetDefinition, ItemAssetsExtension +from pystac.extensions.item_assets import ItemAssetsExtension from pystac.extensions.mgrs import MgrsExtension from pystac.extensions.pointcloud import PointcloudExtension from pystac.extensions.projection import ProjectionExtension @@ -22,8 +30,8 @@ from pystac.extensions.view import ViewExtension from pystac.extensions.xarray_assets import XarrayAssetsExtension -T = TypeVar("T", Asset, AssetDefinition, Link) -U = TypeVar("U", Asset, AssetDefinition) +T = TypeVar("T", Asset, ItemAssetDefinition, Link) +U = TypeVar("U", Asset, ItemAssetDefinition) EXTENSION_NAMES = Literal[ "classification", @@ -107,7 +115,7 @@ def cube(self) -> DatacubeExtension[Collection]: return DatacubeExtension.ext(self.stac_object) @property - def item_assets(self) -> dict[str, AssetDefinition]: + def item_assets(self) -> dict[str, ItemAssetDefinition]: return ItemAssetsExtension.ext(self.stac_object).item_assets @property @@ -300,8 +308,8 @@ def xarray(self) -> XarrayAssetsExtension[Asset]: @dataclass -class ItemAssetExt(_AssetExt[AssetDefinition]): - stac_object: AssetDefinition +class ItemAssetExt(_AssetExt[ItemAssetDefinition]): + stac_object: ItemAssetDefinition @dataclass diff --git a/pystac/extensions/item_assets.py b/pystac/extensions/item_assets.py index 37c3b72d..68ddff7f 100644 --- a/pystac/extensions/item_assets.py +++ b/pystac/extensions/item_assets.py @@ -2,223 +2,24 @@ from __future__ import annotations -from copy import deepcopy from typing import TYPE_CHECKING, Any, Literal import pystac from pystac.extensions.base import ExtensionManagementMixin from pystac.extensions.hooks import ExtensionHooks +from pystac.item_assets import ( + ItemAssetDefinition as AssetDefinition, +) from pystac.serialization.identify import STACJSONDescription, STACVersionID from pystac.utils import get_required if TYPE_CHECKING: - from pystac.extensions.ext import ItemAssetExt + pass SCHEMA_URI = "https://stac-extensions.github.io/item-assets/v1.0.0/schema.json" ITEM_ASSETS_PROP = "item_assets" -ASSET_TITLE_PROP = "title" -ASSET_DESC_PROP = "description" -ASSET_TYPE_PROP = "type" -ASSET_ROLES_PROP = "roles" - - -class AssetDefinition: - """Object that contains details about the datafiles that will be included in member - Items for this Collection. - - See the :stac-ext:`Asset Object ` for details. - """ - - properties: dict[str, Any] - - owner: pystac.Collection | None - - def __init__( - self, properties: dict[str, Any], owner: pystac.Collection | None = None - ) -> None: - self.properties = properties - self.owner = owner - - def __eq__(self, o: object) -> bool: - if not isinstance(o, AssetDefinition): - return NotImplemented - return self.to_dict() == o.to_dict() - - @classmethod - def create( - cls, - title: str | None, - description: str | None, - media_type: str | None, - roles: list[str] | None, - extra_fields: dict[str, Any] | None = None, - ) -> AssetDefinition: - """ - Creates a new asset definition. - - Args: - title : Displayed title for clients and users. - description : Description of the Asset providing additional details, - such as how it was processed or created. - `CommonMark 0.29 `__ syntax MAY be used - for rich text representation. - media_type : `media type\ - `__ - of the asset. - roles : `semantic roles - `__ - of the asset, similar to the use of rel in links. - extra_fields : Additional fields on the asset definition, e.g. from - extensions. - """ - asset_defn = cls({}) - asset_defn.apply( - title=title, - description=description, - media_type=media_type, - roles=roles, - extra_fields=extra_fields, - ) - return asset_defn - - def apply( - self, - title: str | None, - description: str | None, - media_type: str | None, - roles: list[str] | None, - extra_fields: dict[str, Any] | None = None, - ) -> None: - """ - Sets the properties for this asset definition. - - Args: - title : Displayed title for clients and users. - description : Description of the Asset providing additional details, - such as how it was processed or created. - `CommonMark 0.29 `__ syntax MAY be used - for rich text representation. - media_type : `media type\ - `__ - of the asset. - roles : `semantic roles - `__ - of the asset, similar to the use of rel in links. - extra_fields : Additional fields on the asset definition, e.g. from - extensions. - """ - if extra_fields: - self.properties.update(extra_fields) - self.title = title - self.description = description - self.media_type = media_type - self.roles = roles - self.owner = None - - def set_owner(self, obj: pystac.Collection) -> None: - """Sets the owning item of this AssetDefinition. - - The owning item will be used to resolve relative HREFs of this asset. - - Args: - obj: The Collection that owns this asset. - """ - self.owner = obj - - @property - def title(self) -> str | None: - """Gets or sets the displayed title for clients and users.""" - return self.properties.get(ASSET_TITLE_PROP) - - @title.setter - def title(self, v: str | None) -> None: - if v is None: - self.properties.pop(ASSET_TITLE_PROP, None) - else: - self.properties[ASSET_TITLE_PROP] = v - - @property - def description(self) -> str | None: - """Gets or sets a description of the Asset providing additional details, such as - how it was processed or created. `CommonMark 0.29 `__ - syntax MAY be used for rich text representation.""" - return self.properties.get(ASSET_DESC_PROP) - - @description.setter - def description(self, v: str | None) -> None: - if v is None: - self.properties.pop(ASSET_DESC_PROP, None) - else: - self.properties[ASSET_DESC_PROP] = v - - @property - def media_type(self) -> str | None: - """Gets or sets the `media type - `__ - of the asset.""" - return self.properties.get(ASSET_TYPE_PROP) - - @media_type.setter - def media_type(self, v: str | None) -> None: - if v is None: - self.properties.pop(ASSET_TYPE_PROP, None) - else: - self.properties[ASSET_TYPE_PROP] = v - - @property - def roles(self) -> list[str] | None: - """Gets or sets the `semantic roles - `__ - of the asset, similar to the use of rel in links.""" - return self.properties.get(ASSET_ROLES_PROP) - - @roles.setter - def roles(self, v: list[str] | None) -> None: - if v is None: - self.properties.pop(ASSET_ROLES_PROP, None) - else: - self.properties[ASSET_ROLES_PROP] = v - - def to_dict(self) -> dict[str, Any]: - """Returns a dictionary representing this ``AssetDefinition``.""" - return deepcopy(self.properties) - - def create_asset(self, href: str) -> pystac.Asset: - """Creates a new :class:`~pystac.Asset` instance using the fields from this - ``AssetDefinition`` and the given ``href``.""" - return pystac.Asset( - href=href, - title=self.title, - description=self.description, - media_type=self.media_type, - roles=self.roles, - extra_fields={ - k: v - for k, v in self.properties.items() - if k - not in { - ASSET_TITLE_PROP, - ASSET_DESC_PROP, - ASSET_TYPE_PROP, - ASSET_ROLES_PROP, - } - }, - ) - - @property - def ext(self) -> ItemAssetExt: - """Accessor for extension classes on this item_asset - - Example:: - - collection.ext.item_assets["data"].ext.proj.epsg = 4326 - """ - from pystac.extensions.ext import ItemAssetExt - - return ItemAssetExt(stac_object=self) - class ItemAssetsExtension(ExtensionManagementMixin[pystac.Collection]): name: Literal["item_assets"] = "item_assets" diff --git a/pystac/extensions/pointcloud.py b/pystac/extensions/pointcloud.py index b82566e7..804c5537 100644 --- a/pystac/extensions/pointcloud.py +++ b/pystac/extensions/pointcloud.py @@ -13,7 +13,6 @@ ) import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -23,7 +22,7 @@ from pystac.summaries import RangeSummary from pystac.utils import StringEnum, get_required, map_opt -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI: str = "https://stac-extensions.github.io/pointcloud/v1.0.0/schema.json" PREFIX: str = "pc:" @@ -468,7 +467,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> PointcloudExtension[T]: ) cls.ensure_owner_has_extension(obj, add_if_missing) return cast(PointcloudExtension[T], AssetPointcloudExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(PointcloudExtension[T], ItemAssetsPointcloudExtension(obj)) else: @@ -534,11 +533,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsPointcloudExtension(PointcloudExtension[item_assets.AssetDefinition]): +class ItemAssetsPointcloudExtension(PointcloudExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/projection.py b/pystac/extensions/projection.py index 1982e8b5..8d3cb152 100644 --- a/pystac/extensions/projection.py +++ b/pystac/extensions/projection.py @@ -15,7 +15,6 @@ ) import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -23,7 +22,7 @@ ) from pystac.extensions.hooks import ExtensionHooks -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI: str = "https://stac-extensions.github.io/projection/v1.1.0/schema.json" SCHEMA_URIS: list[str] = [ @@ -301,7 +300,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> ProjectionExtension[T]: elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(ProjectionExtension[T], AssetProjectionExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(ProjectionExtension[T], ItemAssetsProjectionExtension(obj)) else: @@ -368,11 +367,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsProjectionExtension(ProjectionExtension[item_assets.AssetDefinition]): +class ItemAssetsProjectionExtension(ProjectionExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/raster.py b/pystac/extensions/raster.py index ab57d7c3..eb01ee6b 100644 --- a/pystac/extensions/raster.py +++ b/pystac/extensions/raster.py @@ -14,7 +14,6 @@ ) import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -23,7 +22,7 @@ from pystac.extensions.hooks import ExtensionHooks from pystac.utils import StringEnum, get_opt, get_required, map_opt -T = TypeVar("T", pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI = "https://stac-extensions.github.io/raster/v1.1.0/schema.json" SCHEMA_URIS = [ @@ -663,7 +662,7 @@ class RasterExtension( ): """An abstract class that can be used to extend the properties of an :class:`~pystac.Item`, :class:`~pystac.Asset`, or - :class:`~pystac.extension.item_assets.AssetDefinition` with properties from + :class:`~pystac.extension.pystac.ItemAssetDefinition` with properties from the :stac-ext:`Raster Extension `. This class is generic over the type of STAC Object to be extended (e.g. :class:`~pystac.Item`, :class:`~pystac.Asset`). @@ -736,7 +735,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> RasterExtension[T]: if isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(RasterExtension[T], AssetRasterExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(RasterExtension[T], ItemAssetsRasterExtension(obj)) else: @@ -771,16 +770,16 @@ def __repr__(self) -> str: return f"" -class ItemAssetsRasterExtension(RasterExtension[item_assets.AssetDefinition]): - asset_definition: item_assets.AssetDefinition - """A reference to the :class:`~pystac.extensions.item_assets.AssetDefinition` +class ItemAssetsRasterExtension(RasterExtension[pystac.ItemAssetDefinition]): + asset_definition: pystac.ItemAssetDefinition + """A reference to the :class:`~pystac.extensions.pystac.ItemAssetDefinition` being extended.""" properties: dict[str, Any] - """The :class:`~pystac.extensions.item_assets.AssetDefinition` fields, including + """The :class:`~pystac.extensions.pystac.ItemAssetDefinition` fields, including extension properties.""" - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.properties = item_asset.properties self.asset_definition = item_asset diff --git a/pystac/extensions/sar.py b/pystac/extensions/sar.py index 112c63a9..16127221 100644 --- a/pystac/extensions/sar.py +++ b/pystac/extensions/sar.py @@ -13,7 +13,6 @@ ) import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -24,7 +23,7 @@ from pystac.summaries import RangeSummary from pystac.utils import StringEnum, get_required, map_opt -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI: str = "https://stac-extensions.github.io/sar/v1.0.0/schema.json" PREFIX: str = "sar:" @@ -332,7 +331,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> SarExtension[T]: ) cls.ensure_owner_has_extension(obj, add_if_missing) return cast(SarExtension[T], AssetSarExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(SarExtension[T], ItemAssetsSarExtension(obj)) else: @@ -399,11 +398,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsSarExtension(SarExtension[item_assets.AssetDefinition]): +class ItemAssetsSarExtension(SarExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/sat.py b/pystac/extensions/sat.py index ab5b638f..ed900b46 100644 --- a/pystac/extensions/sat.py +++ b/pystac/extensions/sat.py @@ -14,7 +14,6 @@ ) import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -24,7 +23,7 @@ from pystac.summaries import RangeSummary from pystac.utils import StringEnum, datetime_to_str, map_opt, str_to_datetime -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI = "https://stac-extensions.github.io/sat/v1.0.0/schema.json" @@ -163,7 +162,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> SatExtension[T]: elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(SatExtension[T], AssetSatExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(SatExtension[T], ItemAssetsSatExtension(obj)) else: @@ -232,11 +231,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsSatExtension(SatExtension[item_assets.AssetDefinition]): +class ItemAssetsSatExtension(SatExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/storage.py b/pystac/extensions/storage.py index 0b0aa0a1..e7155887 100644 --- a/pystac/extensions/storage.py +++ b/pystac/extensions/storage.py @@ -16,7 +16,6 @@ ) import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -25,7 +24,7 @@ from pystac.extensions.hooks import ExtensionHooks from pystac.utils import StringEnum -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI: str = "https://stac-extensions.github.io/storage/v1.0.0/schema.json" PREFIX: str = "storage:" @@ -154,7 +153,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> StorageExtension[T]: elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(StorageExtension[T], AssetStorageExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(StorageExtension[T], ItemAssetsStorageExtension(obj)) else: @@ -221,11 +220,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsStorageExtension(StorageExtension[item_assets.AssetDefinition]): +class ItemAssetsStorageExtension(StorageExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/table.py b/pystac/extensions/table.py index 782d741a..55feda51 100644 --- a/pystac/extensions/table.py +++ b/pystac/extensions/table.py @@ -5,13 +5,12 @@ from typing import Any, Generic, Literal, TypeVar, Union, cast import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ExtensionManagementMixin, PropertiesExtension from pystac.extensions.hooks import ExtensionHooks from pystac.utils import get_required T = TypeVar( - "T", pystac.Collection, pystac.Item, pystac.Asset, item_assets.AssetDefinition + "T", pystac.Collection, pystac.Item, pystac.Asset, pystac.ItemAssetDefinition ) SCHEMA_URI = "https://stac-extensions.github.io/table/v1.2.0/schema.json" @@ -165,7 +164,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> TableExtension[T]: if isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(TableExtension[T], AssetTableExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(TableExtension[T], ItemAssetsTableExtension(obj)) else: @@ -294,11 +293,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsTableExtension(TableExtension[item_assets.AssetDefinition]): +class ItemAssetsTableExtension(TableExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/extensions/version.py b/pystac/extensions/version.py index ee394f48..951a7c95 100644 --- a/pystac/extensions/version.py +++ b/pystac/extensions/version.py @@ -20,6 +20,7 @@ Collection, ExtensionTypeError, Item, + ItemAssetDefinition, Link, MediaType, STACObject, @@ -28,10 +29,9 @@ from pystac.errors import DeprecatedWarning from pystac.extensions.base import ExtensionManagementMixin, PropertiesExtension from pystac.extensions.hooks import ExtensionHooks -from pystac.extensions.item_assets import AssetDefinition from pystac.utils import StringEnum, map_opt -T = TypeVar("T", Collection, Item, Catalog, Asset, AssetDefinition) +T = TypeVar("T", Collection, Item, Catalog, Asset, ItemAssetDefinition) U = TypeVar("U", Collection, Item, Catalog) SCHEMA_URI = "https://stac-extensions.github.io/version/v1.2.0/schema.json" @@ -395,10 +395,10 @@ def __repr__(self) -> str: return f"" -class ItemAssetsViewExtension(BaseVersionExtension[AssetDefinition]): +class ItemAssetsViewExtension(BaseVersionExtension[ItemAssetDefinition]): properties: dict[str, Any] - def __init__(self, item_asset: AssetDefinition): + def __init__(self, item_asset: ItemAssetDefinition): self.properties = item_asset.properties diff --git a/pystac/extensions/view.py b/pystac/extensions/view.py index 3a91249e..29470d87 100644 --- a/pystac/extensions/view.py +++ b/pystac/extensions/view.py @@ -6,7 +6,6 @@ from typing import Any, Generic, Literal, TypeVar, Union, cast import pystac -from pystac.extensions import item_assets from pystac.extensions.base import ( ExtensionManagementMixin, PropertiesExtension, @@ -15,7 +14,7 @@ from pystac.extensions.hooks import ExtensionHooks from pystac.summaries import RangeSummary -T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition) +T = TypeVar("T", pystac.Item, pystac.Asset, pystac.ItemAssetDefinition) SCHEMA_URI: str = "https://stac-extensions.github.io/view/v1.0.0/schema.json" PREFIX: str = "view:" @@ -166,7 +165,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> ViewExtension[T]: elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(ViewExtension[T], AssetViewExtension(obj)) - elif isinstance(obj, item_assets.AssetDefinition): + elif isinstance(obj, pystac.ItemAssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(ViewExtension[T], ItemAssetsViewExtension(obj)) else: @@ -233,11 +232,11 @@ def __repr__(self) -> str: return f"" -class ItemAssetsViewExtension(ViewExtension[item_assets.AssetDefinition]): +class ItemAssetsViewExtension(ViewExtension[pystac.ItemAssetDefinition]): properties: dict[str, Any] - asset_defn: item_assets.AssetDefinition + asset_defn: pystac.ItemAssetDefinition - def __init__(self, item_asset: item_assets.AssetDefinition): + def __init__(self, item_asset: pystac.ItemAssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties diff --git a/pystac/item_assets.py b/pystac/item_assets.py new file mode 100644 index 00000000..4bcaec9d --- /dev/null +++ b/pystac/item_assets.py @@ -0,0 +1,243 @@ +"""Implements the ``Item Asset Definition ``.""" + +from __future__ import annotations + +from copy import deepcopy +from typing import TYPE_CHECKING, Any + +import pystac + +if TYPE_CHECKING: + from pystac.extensions.ext import ItemAssetExt + + +ASSET_TITLE_PROP = "title" +ASSET_DESC_PROP = "description" +ASSET_TYPE_PROP = "type" +ASSET_ROLES_PROP = "roles" + + +class ItemAssetDefinition: + """Object that contains details about the datafiles that will be included in member + Items for this Collection. + + See the `Item Asset Definition Object ` for details. + """ + + properties: dict[str, Any] + + owner: pystac.Collection | None + + def __init__( + self, properties: dict[str, Any], owner: pystac.Collection | None = None + ) -> None: + self.properties = properties + self.owner = owner + + def __eq__(self, o: object) -> bool: + if not isinstance(o, ItemAssetDefinition): + return NotImplemented + return self.to_dict() == o.to_dict() + + @classmethod + def create( + cls, + title: str | None, + description: str | None, + media_type: str | None, + roles: list[str] | None, + extra_fields: dict[str, Any] | None = None, + ) -> ItemAssetDefinition: + """ + Creates a new asset definition. + + Args: + title : Displayed title for clients and users. + description : Description of the Asset providing additional details, + such as how it was processed or created. + `CommonMark 0.29 `__ syntax MAY be used + for rich text representation. + media_type : `media type\ + `__ + of the asset. + roles : `semantic roles + `__ + of the asset, similar to the use of rel in links. + extra_fields : Additional fields on the asset definition, e.g. from + extensions. + """ + asset_defn = cls({}) + asset_defn.apply( + title=title, + description=description, + media_type=media_type, + roles=roles, + extra_fields=extra_fields, + ) + return asset_defn + + def apply( + self, + title: str | None, + description: str | None, + media_type: str | None, + roles: list[str] | None, + extra_fields: dict[str, Any] | None = None, + ) -> None: + """ + Sets the properties for this asset definition. + + Args: + title : Displayed title for clients and users. + description : Description of the Asset providing additional details, + such as how it was processed or created. + `CommonMark 0.29 `__ syntax MAY be used + for rich text representation. + media_type : `media type\ + `__ + of the asset. + roles : `semantic roles + `__ + of the asset, similar to the use of rel in links. + extra_fields : Additional fields on the asset definition, e.g. from + extensions. + """ + if extra_fields: + self.properties.update(extra_fields) + self.title = title + self.description = description + self.media_type = media_type + self.roles = roles + self.owner = None + + def set_owner(self, obj: pystac.Collection) -> None: + """Sets the owning item of this ItemAssetDefinition. + + The owning item will be used to resolve relative HREFs of this asset. + + Args: + obj: The Collection that owns this asset. + """ + self.owner = obj + + @property + def title(self) -> str | None: + """Gets or sets the displayed title for clients and users.""" + return self.properties.get(ASSET_TITLE_PROP) + + @title.setter + def title(self, v: str | None) -> None: + if v is None: + self.properties.pop(ASSET_TITLE_PROP, None) + else: + self.properties[ASSET_TITLE_PROP] = v + + @property + def description(self) -> str | None: + """Gets or sets a description of the Asset providing additional details, such as + how it was processed or created. `CommonMark 0.29 `__ + syntax MAY be used for rich text representation.""" + return self.properties.get(ASSET_DESC_PROP) + + @description.setter + def description(self, v: str | None) -> None: + if v is None: + self.properties.pop(ASSET_DESC_PROP, None) + else: + self.properties[ASSET_DESC_PROP] = v + + @property + def media_type(self) -> str | None: + """Gets or sets the `media type + `__ + of the asset.""" + return self.properties.get(ASSET_TYPE_PROP) + + @media_type.setter + def media_type(self, v: str | None) -> None: + if v is None: + self.properties.pop(ASSET_TYPE_PROP, None) + else: + self.properties[ASSET_TYPE_PROP] = v + + @property + def roles(self) -> list[str] | None: + """Gets or sets the `semantic roles + `__ + of the asset, similar to the use of rel in links.""" + return self.properties.get(ASSET_ROLES_PROP) + + @roles.setter + def roles(self, v: list[str] | None) -> None: + if v is None: + self.properties.pop(ASSET_ROLES_PROP, None) + else: + self.properties[ASSET_ROLES_PROP] = v + + def to_dict(self) -> dict[str, Any]: + """Returns a dictionary representing this ``ItemAssetDefinition``.""" + return deepcopy(self.properties) + + def create_asset(self, href: str) -> pystac.Asset: + """Creates a new :class:`~pystac.Asset` instance using the fields from this + ``ItemAssetDefinition`` and the given ``href``.""" + return pystac.Asset( + href=href, + title=self.title, + description=self.description, + media_type=self.media_type, + roles=self.roles, + extra_fields={ + k: v + for k, v in self.properties.items() + if k + not in { + ASSET_TITLE_PROP, + ASSET_DESC_PROP, + ASSET_TYPE_PROP, + ASSET_ROLES_PROP, + } + }, + ) + + @property + def ext(self) -> ItemAssetExt: + """Accessor for extension classes on this item_asset + + Example:: + + collection.item_assets["data"].ext.proj.epsg = 4326 + """ + from pystac.extensions.ext import ItemAssetExt + + return ItemAssetExt(stac_object=self) + + +class _ItemAssets(dict): # type:ignore + """Private class for exposing item_assets as a dict + + This class coerces values to ``ItemAssetDefinition``s and + sets that owner on all ``ItemAssetDefinition``s to the collection + that it is owned by. + """ + + collection: pystac.Collection + + def __init__(self, collection: pystac.Collection) -> None: + self.collection = collection + if not collection.extra_fields.get("item_assets"): + collection.extra_fields["item_assets"] = {} + self.update(collection.extra_fields["item_assets"]) + + def __setitem__(self, key: str, value: Any) -> None: + if isinstance(value, ItemAssetDefinition): + asset_definition = value + asset_definition.set_owner(self.collection) + else: + asset_definition = ItemAssetDefinition(value, self.collection) + self.collection.extra_fields["item_assets"][key] = asset_definition.properties + super().__setitem__(key, asset_definition) + + def update(self, *args: Any, **kwargs: Any) -> None: + for k, v in dict(*args, **kwargs).items(): + self[k] = v