Copyright (C) 2020-2022 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
HERE platform, content adaptation, encoding and decoding to standard
Python classes such as list, dictionaries, and protocol buffers Message types.
"""
import csv
import json
import tempfile
from collections import defaultdict
from io import StringIO
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
Iterator,
List,
Mapping,
Optional,
Tuple,
Union,
cast,
)

import geojson
from geojson import Feature, FeatureCollection
from geojson.mapping import GEO_INTERFACE_MARKER
from google.protobuf.message import Message
from here.platform.adapter import Adapter, ContentAdapter, Decoder, Encoder, Ref
from here.platform.partition import (
IndexPartition,
StreamPartition,
VersionedPartition,
VolatilePartition,
)
from here.platform.schema import Schema
from here.platform.utils.collection import iter_tuples

if TYPE_CHECKING:
from here.platform.layer import StreamLayer, VersionedLayer, VolatileLayer

from here.platform.adapter import Identifier, Partition

[docs]
class DefaultEncoder(Encoder):
"""A content encoder and its capabilities."""

@property
def supported_content_types(self) -> Dict[str, Union[type, Tuple[type, ...]]]:
"""
:return: the dictionary of MIME content types supported when encoding single blobs
with the encode_blob function of this encoder, each with the type of the encoded data.
"""
return {"application/x-protobuf": Message, "application/json": dict, "text/csv": list, "application/vnd.geo+json": (dict, FeatureCollection), "application/geo+json": (dict, FeatureCollection),}

[docs]
def encode_blob(
self, data, content_type: str, schema: Optional[Schema] = None, **kwargs
) -> bytes:
"""Encode one single blob of data.

:param data: the data to be encoded, its type corresponds to the type declared in
the property supported_content_types for the content type
:param content_type: the MIME content type to be encoded
:param schema: the schema, if the content type requires one
:param kwargs: additional, content-type-specific parameters for the encoder:

For JSON (application/json):
For additional info, please see:
https://docs.python.org/3/library/json.html?highlight=json#json.dumps

For CSV (text/csv):
For additional info, please see:
https://docs.python.org/3/library/csv.html#csv.DictWriter

For GeoJSON (application/geo+json or application/vnd.geo+json):
For additional info, please see:
https://github.com/jazzband/geojson#geojson-encoding-decoding

:return: the encoded data
:raises UnsupportedContentTypeEncodeException: in case the content type is not encodable
:raises ValueError: if the schema is mandatory for the content type but missing
:raises EncodeException: in case the blob can't be properly encoded # noqa
:raises SchemaException: in case the schema can't be used to encode the content # noqa
"""
self._verify_encode_blob(data, content_type, schema)
encoded = None

if content_type == "application/json":
encoded = (
schema.encode_blob(data)
if schema
else DefaultEncoder._encode_json_blob(data, **kwargs)
)

if content_type == "text/csv":
encoded = DefaultEncoder._encode_csv_blob(data, **kwargs)

elif content_type in ("application/geo+json", "application/vnd.geo+json"):
encoded = (
schema.encode_blob(data)
if schema
else DefaultEncoder._encode_geojson_blob(data, **kwargs)
)

elif content_type == "application/x-protobuf":
if schema:
encoded = schema.encode_blob(data)
else:
raise ValueError(f"Schema is required to encode content type {content_type}")

assert encoded is not None # if this fails, it's because of supported_content_types
return encoded

@staticmethod
def _encode_json_blob(data: dict, **kwargs) -> bytes:
data_bytes: bytes = json.dumps(data, **kwargs).encode("utf-8")
return data_bytes

@staticmethod
def _encode_csv_blob(data: list, **kwargs) -> bytes:
with tempfile.NamedTemporaryFile(mode="r+") as tmp:
writer = csv.DictWriter(tmp, **kwargs)
writer.writeheader()
writer.writerows(data)
tmp.seek(0)
return tmp.read().encode("utf-8")

@staticmethod
def _encode_geojson_blob(data: dict, **kwargs) -> bytes:
data_bytes: bytes = geojson.dumps(data, **kwargs).encode("utf-8")
return data_bytes

[docs]
class DefaultDecoder(Decoder):
"""
The decoder provided by default in case no Adapter is configured for the platform
or for a specific catalog.
"""

@property
def supported_content_types(self) -> Dict[str, Union[type, Tuple[type, ...]]]:
"""
:return: the dictionary of MIME content types supported when decoding single blobs
with the decode_blob function of this decoder, each with the type of the decoded data.
"""
return {"application/protobuf": Message, "application/x-protobuf": Message, "application/json": dict, "text/csv": list, "application/vnd.geo+json": FeatureCollection, "application/geo+json": FeatureCollection,}

[docs]
def decode_blob(
self, data: bytes, content_type: str, schema: Optional[Schema] = None, **kwargs
):
"""Decode one single blob of data.

:param data: the encoded data
:param content_type: the MIME content type to be decoded
:param schema: the schema, if the content type requires one
:param kwargs: additional, content-type-specific parameters for the decoder:

For JSON (application/json):
For additional info, please see:
https://docs.python.org/3/library/json.html?highlight=json#json.loads

For CSV (text/csv):
For additional info, please see:
https://docs.python.org/3/library/csv.html#csv.DictReader

For GeoJSON (application/geo+json or application/vnd.geo+json):
For additional info, please see:
https://github.com/jazzband/geojson#geojson-encoding-decoding

:return: the decoded blob, its type correspond to the type declared in
the property decodable_content_types for the content type
:raises UnsupportedContentTypeDecodeException: in case the content type is not decodable
:raises ValueError: if the schema is mandatory for the content type but missing
:raises DecodeException: in case the blob can't be properly decoded # noqa
:raises SchemaException: in case the schema can't be used to decode the content # noqa
"""
self._verify_decode_blob(data, content_type, schema)
decoded: Any = None
if content_type in ["application/protobuf", "application/x-protobuf"]:
if not schema:
raise ValueError(f"Schema is required to decode content type {content_type}")

Actual protobuf decoding

decoded = schema.decode_blob(data)

elif content_type == "application/json":
decoded = (
schema.decode_blob(data)
if schema
else DefaultDecoder._decode_json_blob(data, **kwargs)
)

elif content_type in ["application/vnd.geo+json", "application/geo+json"]:
decoded = (
schema.decode_blob(data)
if schema
else DefaultDecoder._decode_geojson_blob(data, **kwargs)
)

elif content_type == "text/csv":
decoded = DefaultDecoder._decode_csv_blob(data, **kwargs)

assert decoded is not None # if this fails, it's because of supported_content_types
return self._verify_and_return_decoded(decoded, content_type)

@staticmethod
def _decode_json_blob(blob: bytes, **kwargs) -> dict:
dict_obj: dict = json.loads(blob, **kwargs)
return dict_obj

@staticmethod
def _decode_csv_blob(blob: bytes, **kwargs) -> list:
content = blob.decode("utf-8")
file = StringIO(content)
return list(csv.DictReader(file, **kwargs))

@staticmethod
def _decode_geojson_blob(blob: bytes, **kwargs) -> FeatureCollection:
feature_collection = geojson.loads(blob, **kwargs)
if not isinstance(feature_collection, FeatureCollection):
raise ValueError("Blob does not contain a valid GeoJSON FeatureCollection")
return feature_collection

[docs]
class DefaultContentAdapter(ContentAdapter):
"""The default content adapter, that exposes content via dictionaries and lists."""

[docs]
def from_tabular(self, columns, data, geometry_column="geometry"):
"""
Convert the given tabular data.

:param columns: column names
:param data: tabular data
:param geometry_column: geometry column string
:return: data representation is function of the adapter # noqa
:raises NotImplementedError: so far.
"""
raise NotImplementedError

[docs]
def from_objects(
self,
fields: type,
data: Iterator[object],
single_element: bool = False,
index_partition: Union[None, str, Callable[[object], Partition]] = None,
index_id: Union[None, str, Callable[[object], Identifier]] = None,
index_ref: Union[None, str, Callable[[object], Union[Ref, Iterable[Ref]]]] = None,
) -> Union[
object, # one single object
List[object], # non-indexed objects
Dict[Partition, object], # objects indexed by partition
Dict[Identifier, object], # objects indexed by id
Dict[Partition, Dict[Identifier, object]], # objects indexed by partition and id
Dict[Ref, List[object]], # objects indexed by reference
]:
"""
Adapt content form an object representation to the target format.

It can optionally perform indexing of objects, based on their partition,
identifier and set of references to other objects. Indexing is specified by
naming the field of the object that contains the value to index, or by
passing a function that calculates that value from the object.

If indexing by object id is enabled, object are indexed by their identifier,
nesting them into an outer dictionary with the object identifier used as key:

{o1_id: o1, o2_id: o2, o3_id: o3, ...}

When both partition and object indexing is enabled, object are indexed by partition
identifier, nesting them into an outer dictionary with the partition identifiers used
as key. This wraps also the dictionary that index objects by their identifiers:

{p1_id: { o1_id: o1, o2_id: o2, ... }, p2_id: { o3_id: o3, ... }}

When only indexing by partition is enabled, objects are returned in one single dict:
{p1_id: [ o1, o2, ... ], p2_id: [ o3, ... ]}

When no indexing is enabled, objects are returned in a list:
[ o1, o2, o3, ... ]

When indexing by reference is enabled, objects are grouped and indexed
by the zero, one or more reference they contain. Result is indexed by reference.
This may result in the same object present more than once:

{r1: [ o1, o3, ... ], r2: [ o2, o3, ... ], r3: [ o1, o2, o4, ... ]}

:param fields: the fields to extract, as specified by a dataclass. Field names are looked
up among the attributes of each object via getattr`. When missing, Noneor equivalent is used. Each field has a type that describes its semantic: it is used to adapt the value to the most appropriate representation for the output format.TypeErroris raised in case this is not possible. :param data: the objects to adapt to the target format. Fields not mentioned infields are discarded. Expected but missing fields and identifiers are consideredNone. Field values may be of any type compatible with the type declared for the field. Partition ids don't have to be unique, but they have to be contiguous: all the objects with a given partition identifier must be returned in sequence. Object identifiers, when present, must be unique across the whole content. :param single_element: the data contains exactly one element, the content adapter case use this information to optimize or return a specialized representation :param index_partition: index the content by partition, using the field specified :param index_id: index the content by object identifier, using the field specified :param index_ref: index the content by references, using the field specified. Each object can contain zero, one or more references, and references can be shared among multiple objects. :return: objects indexed as requested. Indexing is implemented with Dict``.
:raises:
ValueError: if the fields are not described by a dataclass
KeyError: in case partition id, object id or reference is needed but not present
TypeError: in case partition or object id is not of type int or string.
Also raised in case field values are not of the type declared for the field,
or if they can't be converted to it.
"""

Function main code begins here

ContentAdapter._validate_params(single_element, index_partition, index_id, index_ref)
ContentAdapter._validate_fields(fields)
ContentAdapter._validate_indices(fields, index_partition, index_id)

Enable on-the-fly validation of data

data = map(lambda x: ContentAdapter._validate_object(fields, x), data)

Single element representation is obtained by simply returning

the first object unchanged without any indexing structure around

if single_element:
return next(data)

if index_ref:
result: Dict[Ref, List[object]] = defaultdict(list)
for obj in data:
for ref in ContentAdapter._extract_refs(index_ref, obj):
result[ref].append(obj)
return result
elif index_partition and index_id:
result2: Dict[Partition, Dict[Identifier, object]] = defaultdict(dict)
for obj in data:
partition = ContentAdapter._extract_partition(index_partition, obj)
id = ContentAdapter._extract_identifier(index_id, obj)
result2[partition][id] = obj
return result2
elif index_partition: # and not index_id
result3: Dict[Partition, List[object]] = defaultdict(list)
for obj in data:
partition = ContentAdapter._extract_partition(index_partition, obj)
result3[partition].append(obj)
return result3
elif index_id: # and not index_partition
return {ContentAdapter._extract_identifier(index_id, obj): obj for obj in data}
else:
return list(data)

[docs]
class DefaultAdapter(Adapter):
"""
This adapter transforms data from and to standard Python classes,
such as list, dictionaries, and protocol buffers Message types.

An adapter controls the encoding and decoding process of platform data.
It transforms data from and to adapter-specific data structure and supports
reading, writing, encoding and decoding a variety of MIME content types.

For the list of MIME content types supported when reading and writing
a layer with read_* and write_* functions of the :class:Layer
and its subclasses, please see documentation of :class:DefaultDecoder
and :class:DefaultEncoder.

All the operations involving content passes through an adapter when
the parameters encode or decode are True, their default value.
These are parameters of the read_* and write_* functions.
If a content type is not supported, or if reading or writing raw content is preferred,
pass False to skip encoding or decoding and deal with raw bytes instead.
"""

def init(self):
"""
Initialize the default adapter.
"""
self._encoder: Encoder = DefaultEncoder()
self._decoder: Decoder = DefaultDecoder()
self._content_adapter: ContentAdapter = DefaultContentAdapter()

@property
def encoder(self) -> Encoder:
"""The encoder associated with the adapter."""
return self._encoder

@property
def decoder(self) -> Decoder:
"""The decoder associated with the adapter."""
return self._decoder

@property
def content_adapter(self) -> ContentAdapter:
"""The adapter specialized for content."""
return self._content_adapter

[docs]
def from_versioned_metadata(
self, partitions: Iterator[VersionedPartition], **kwargs
) -> Iterator[VersionedPartition]:
"""
Adapt versioned partition metadata to a sequence of :class:VersionedPartition.

:param partitions: sequence of partition metadata from a versioned layer
:param kwargs: unused
:return: partition metadata as sequence of :class:VersionedPartition
"""
return partitions

[docs]
def from_versioned_data(
self,
partitions_data: Iterator[Tuple[VersionedPartition, bytes]],
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> Iterator[Tuple[VersionedPartition, Any]]:
"""Adapt versioned partition metadata and data to a sequence of
class:VersionedPartition and decoded data.

:param partitions_data: sequence of partition metadata and data from a versioned layer
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param kwargs: additional, content-type-specific parameters, see :class:DefaultDecoder
:return: sequence of :class:VersionedPartition and decoded data
"""

def decode(p_data) -> Tuple[VersionedPartition, Any]:
p, data = p_data
return p, self.decoder.decode_blob(data, content_type, schema, **kwargs)

return map(decode, partitions_data)

[docs]
def to_versioned_metadata(
self,
layer: "VersionedLayer",
partitions_update: Optional[Iterator[VersionedPartition]],
partitions_delete: Optional[Iterator[Union[str, int]]],
**kwargs,
) -> Tuple[Iterator[VersionedPartition], Iterator[Union[str, int]]]:
"""Adapt sequences of :class:VersionedPartition and partition ids
to versioned partition metadata and partition ids to update and delete.

:param layer: the layer all the metadata and data belong to
:param partitions_update: the sequence of partitions metadata to update, if any
:param partitions_delete: the sequence of partitions ids to delete, if any
:param kwargs: unused
:return: tuple of Iterator, the first with the :class:VersionedPartition
that have to be updated, the second with the partition ids to delete
"""
return partitions_update or iter([]), partitions_delete or iter([])

[docs]
def to_versioned_data(
self,
layer: "VersionedLayer",
data: Union[Iterator[Tuple[Union[str, int], Any]], Mapping[Union[str, int], Any]],
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> Iterator[Tuple[Union[str, int], bytes]]:
"""Adapt data from sequence of partition ids and data to versioned partition id and data.

:param layer: the layer all the metadata and data belong to
:param data: dictionary or sequence of partition id and content to adapt
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param kwargs: additional, content-type-specific parameters, see :class:DefaultEncoder
:return: sequence of partition id and data for the versioned layer
"""

def encode(pid_msg):
pid, msg = pid_msg
return pid, self.encoder.encode_blob(msg, content_type, schema, **kwargs)

return map(encode, iter_tuples(data))

[docs]
def from_volatile_metadata(
self, partitions: Iterator[VolatilePartition], **kwargs
) -> Iterator[VolatilePartition]:
"""
Adapt volatile partition metadata to a sequence of :class:VolatilePartition.

:param partitions: sequence of partition metadata from a volatile layer
:param kwargs: unused
:return: partition metadata as sequence of :class:VolatilePartition
"""
return partitions

[docs]
def from_volatile_data(
self,
partitions_data: Iterator[Tuple[VolatilePartition, bytes]],
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> Iterator[Tuple[VolatilePartition, Any]]:
"""Adapt volatile partition metadata and data to a sequence of
class:VolatilePartition and decoded data.

:param partitions_data: sequence of partition metadata and data from a volatile layer
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param kwargs: additional, content-type-specific parameters, see :class:DefaultDecoder
:return: sequence of :class:VolatilePartition and decoded data
"""

def decode(p_data) -> Tuple[VolatilePartition, Any]:
p, data = p_data
return p, self.decoder.decode_blob(data, content_type, schema, **kwargs)

return map(decode, partitions_data)

[docs]
def to_volatile_metadata(
self,
layer: "VolatileLayer",
partitions_update: Optional[Iterator[VolatilePartition]],
partitions_delete: Optional[Iterator[Union[str, int]]],
**kwargs,
) -> Tuple[Iterator[VolatilePartition], Iterator[Union[str, int]]]:
"""Adapt sequences of :class:VolatilePartition and partition ids
to volatile partition metadata and partition ids to update and delete.

:param layer: the layer all the metadata and data belong to
:param partitions_update: the sequence of partitions metadata to update, if any
:param partitions_delete: the sequence of partitions ids to delete, if any
:param kwargs: unused
:return: tuple of Iterator, the first with the :class:VolatilePartition
that have to be updated, the second with the partition ids to delete
"""
return partitions_update or iter([]), partitions_delete or iter([])

[docs]
def to_volatile_data(
self,
layer: "VolatileLayer",
data: Union[Iterator[Tuple[Union[str, int], Any]], Mapping[Union[str, int], Any]],
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> Iterator[Tuple[Union[str, int], bytes]]:
"""Adapt data from sequence of partition ids and data to volatile partition id and data.

:param layer: the layer all the metadata and data belong to
:param data: dictionary or sequence of partition id and content to adapt
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param kwargs: additional, content-type-specific parameters, see :class:DefaultEncoder
:return: sequence of partition id and data for the volatile layer
"""

def encode(pid_msg):
pid, msg = pid_msg
return pid, self.encoder.encode_blob(msg, content_type, schema, **kwargs)

return map(encode, iter_tuples(data))

[docs]
def from_stream_metadata(
self, partitions: Iterator[StreamPartition], **kwargs
) -> Iterator[StreamPartition]:
"""Adapt stream partition metadata to a sequence of :class:StreamPartition.

:param partitions: sequence of partition metadata from a stream layer
:param kwargs: unused
:return: partition metadata as sequence of :class:StreamPartition
"""
return partitions

[docs]
def from_stream_data(
self,
partitions_data: Iterator[Tuple[StreamPartition, bytes]],
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> Iterator[Tuple[StreamPartition, Any]]:
"""Adapt versioned partition metadata and data to a sequence of
class:StreamPartition and decoded data.

:param partitions_data: sequence of partition metadata and data from a stream layer
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param kwargs: additional, content-type-specific parameters, see :class:DefaultDecoder
:return: sequence of :class:StreamPartition and decoded data
"""

def decode(p_data) -> Tuple[StreamPartition, Any]:
p, data = p_data
return p, self.decoder.decode_blob(data, content_type, schema, **kwargs)

return map(decode, partitions_data)

[docs]
def to_stream_metadata(
self, layer: "StreamLayer", partitions: Iterator[StreamPartition], **kwargs
) -> Iterator[StreamPartition]:
"""Adapt what to publish from the target format to stream partition metadata.

:param layer: the layer all the metadata and data belong to
:param partitions: sequence of partitions metadata to publish
:param kwargs: unused
:return: Iterator with the :class:StreamPartition that are adapted
"""
return partitions

[docs]
def to_stream_data(
self,
layer: "StreamLayer",
data: Union[
Iterable[
Union[
Tuple[Union[str, int], Any], # id, data
Tuple[Union[str, int], Any, Optional[int]], # id, data, ts
]
],
Mapping[Union[str, int], Any], # id, data
],
content_type: str,
schema: Optional[Schema],
timestamp: Optional[int],
**kwargs,
) -> Iterator[Tuple[Union[str, int], bytes, Optional[int]]]:
"""Adapt data from the target format to stream partition metadata and data.

:param layer: the layer all the metadata and data belong to
:param data: sequence of partition id, content and optional timestamp to adapt
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param timestamp: optional timestamp for all the messages, if none is specified in data:
in milliseconds since Unix epoch (1970-01-01T00:00:00 UTC)
:param kwargs: adapter-specific, please consult the documentation
of the specific adapter to for the parameters and types it supports
:return: sequence of partition id, data and timestamp for the stream layer
"""

def normalize(
entry: Union[
Tuple[Union[str, int], Any], # id, data
Tuple[Union[str, int], Any, Optional[int]], # id, data, ts
]
) -> Tuple[Union[str, int], Any, Optional[int]]:
if len(entry) == 3:
id, data, ts = cast(Tuple[Union[str, int], Any, Optional[int]], entry)
return id, data, (ts or timestamp)
elif len(entry) == 2:
id, data = cast(Tuple[Union[str, int], Any], entry)
return id, data, timestamp
else:
raise ValueError("Unexpected format for data")

def encode(pid_msg_ts):
pid, msg, ts = pid_msg_ts
return pid, self.encoder.encode_blob(msg, content_type, schema, **kwargs), ts

return map(encode, map(normalize, iter_tuples(data)))

[docs]
def from_index_metadata(
self, partitions: Iterator[IndexPartition], **kwargs
) -> Iterator[IndexPartition]:
"""
Adapt index partition metadata to a sequence of :class:IndexPartition.

:param partitions: sequence of partition metadata from an index layer
:param kwargs: unused
:return: partition metadata as sequence of :class:IndexPartition
"""
return partitions

[docs]
def from_index_data(
self,
partitions_data: Iterator[Tuple[IndexPartition, bytes]],
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> Iterator[Tuple[IndexPartition, Any]]:
"""Adapt versioned partition metadata and data to a sequence of
class:IndexPartition and decoded data.

:param partitions_data: sequence of partition metadata and data from an index layer
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param kwargs: additional, content-type-specific parameters, see :class:DefaultDecoder
:return: sequence of :class:IndexPartition and decoded data
"""

def decode(p_data) -> Tuple[IndexPartition, Any]:
p, data = p_data
return p, self.decoder.decode_blob(data, content_type, schema, **kwargs)

return map(decode, partitions_data)

[docs]
def to_index_single_data(
self,
data: None,
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> bytes:
"""Adapt data to be stored in an index layer.

:param data: data (nothing supported so far)
:param content_type: the MIME content type of the layer
:param schema: optional :class:Schema of the layer
:param kwargs: additional, content-type-specific parameters, see :class:DefaultEncoder
:return: data encoded for an index layer
:raises ValueError: in case the content type is not supported by the adapter # noqa
"""
return self.encoder.encode_blob(data, content_type, schema, **kwargs)

[docs]
def from_feature_ids(self, feature_ids: Iterator[str], **kwargs) -> List[str]:
"""Adapt a sequence of feature identifiers to a list.

:param feature_ids: sequence of feature identifiers
:param kwargs: unused
:return: a list of feature identifiers
"""
return list(feature_ids)

[docs]
def to_feature_ids(self, data: Iterable[str], **kwargs) -> Iterator[str]:
"""Adapt data to a sequence of feature identifiers.

:param data: sequence, list or iterator of feature identifiers
:param kwargs: unused
:return: sequence of feature identifiers
"""
return iter(data)

[docs]
def from_geo_features(self, features: Iterator[Feature], **kwargs) -> FeatureCollection:
"""Adapt a sequence of geographic features to a GeoJSON FeatureCollection.

:param features: sequence of geographic features
:param kwargs: unused
:return: a single GeoJSON FeatureCollection containing the features
"""
return FeatureCollection(features=list(features))

[docs]
def to_geo_features(
self,
data: Union[geojson.FeatureCollection, dict, Any, Iterable[Union[Feature, dict, Any]]],
**kwargs,
) -> Iterator[Feature]:
"""Adapt data in a supported format to a sequence of geographic features.

This adapter supports the following formats:

• A :class:CollectionFeature or GeoJSON FeatureCollection dictionary or
  any object implementing the __geo_interface__ with type="FeatureCollection"
  as described in https://gist.github.com/sgillies/2217756
• An iterable of :class:Feature or GeoJSON Feature dictionary or
  any object implementing the __geo_interface__ with type="Feature"
  as described in https://gist.github.com/sgillies/2217756

:param data: a collection of features in one of the supported formats
:param kwargs: unused
:return: sequence of geographic features
:raises ValueError: if data doesn't match any of the supported formats
"""

def to_feature(f):
if isinstance(f, Feature):
return f
if hasattr(f, GEO_INTERFACE_MARKER):
f = getattr(f, GEO_INTERFACE_MARKER)
if isinstance(f, dict):
if f.get("type") != "Feature":
raise ValueError(f"{f} is not a geographic feature")
return Feature(
id=f.get("id"), geometry=f.get("geometry"), properties=f.get("properties")
)
else:
raise ValueError(f"{f} is not a geographic feature")

if hasattr(data, GEO_INTERFACE_MARKER):
data = getattr(data, GEO_INTERFACE_MARKER)
if isinstance(data, dict):
if data.get("type") != "FeatureCollection":
raise ValueError(f"{data} is not a collection of geographic features")
return map(to_feature, data.get("features", []))
else:
return map(to_feature, data)