Source code for here.geopandas_adapter.geopandas_adapter

Copyright (C) 2019-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 Python SDK, GeoPandas adapter access package
"""
import dataclasses
import io
import json
import logging
import typing
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
Iterator,
NamedTuple,
Optional,
Tuple,
Union,
)

import geojson
import geopandas as gpd
import pandas as pd
from geojson import Feature
from google.protobuf.json_format import MessageToDict
from google.protobuf.message import Message
from here.geopandas_adapter.utils.geo import to_geometry
from here.platform.adapter import (
Adapter,
ContentAdapter,
DecodedMessage,
Decoder,
Encoder,
Identifier,
Partition,
Ref,
)
from here.platform.partition import (
IndexPartition,
StreamPartition,
VersionedPartition,
VolatilePartition,
)
from here.platform.schema import ProtobufParser, Schema
from here.platform.utils.collection import flatten_iterator
from pandas import notna
from pyarrow import ArrowNotImplementedError
from shapely import wkt
from shapely.geometry import LineString, Point, shape

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

logger = logging.getLogger(name)

def _extract(named_tuple: NamedTuple, column: str, convert_to):
"""
Take a named tuple (typically coming from pandas itertuples),
extract the value of a column given its name and enforce/convert its type

:param named_tuple: the named tuple to inspect
:param column: name of the column or tuple field to extract
:param convert_to: function to convert the value to the target type,
this can be str, int or any other function, like lambda
:returns: the converted value of the column, if any, None otherwise
"""
if hasattr(named_tuple, column):
value = getattr(named_tuple, column)
return convert_to(value) if notna(value) else None
else:
return None

[docs]
class GeoPandasEncoder(Encoder):
"""
Implementation of an :class:Encoder to work with pd.DataFrame and gpd.GeoDataFrame.
"""

@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/x-parquet": pd.DataFrame, "text/csv": pd.DataFrame, "application/json": pd.DataFrame, "application/vnd.geo+json": gpd.GeoDataFrame, "application/geo+json": gpd.GeoDataFrame,}

@staticmethod
def _encode_parquet_blob(data: pd.DataFrame, engine, **kwargs) -> bytes:
data_bytes = io.BytesIO()
supported_engines = ["auto", "fastparquet", "pyarrow"]
if engine not in supported_engines:
raise ValueError(
f"The engine name is incorrect, available values are {supported_engines}"
)
if engine == "auto":
try:
data.to_parquet(data_bytes, engine="auto", **kwargs)
except ArrowNotImplementedError:
logger.debug("Error in serializing with pyarrow. Trying fastparquet.")
data.to_parquet(data_bytes, engine="fastparquet", **kwargs)
else:
data.to_parquet(data_bytes, engine=engine, **kwargs)

return data_bytes.getvalue()

@staticmethod
def _encode_json_blob(data: pd.DataFrame, **kwargs) -> bytes:
data_bytes: bytes = data.to_json(**kwargs).encode("utf-8")
return data_bytes

@staticmethod
def _encode_geojson_blob(gdf: gpd.GeoDataFrame, **kwargs) -> bytes:
data_bytes: bytes = gdf.to_json(**kwargs).encode("utf-8")
return data_bytes

@staticmethod
def _encode_csv_blob(data: pd.DataFrame, **kwargs) -> bytes:
data_bytes: bytes = data.to_csv(**kwargs).encode("utf-8")
return data_bytes

@staticmethod
def _convert_to_message(key, value, output_message, nesting_level=0, max_nesting_level=20):
if nesting_level > max_nesting_level:
raise ValueError("Max nested level reached")

def set_attr(set_key, set_value, set_message):
try:
setattr(set_message, set_key, set_value)
except TypeError:
setattr(set_message, set_key, int(set_value))

def handle_item(item_key, item_value, parent_message, level):
if isinstance(item_value, list):
for item in item_value:
handle_item(item_key, item, parent_message, level + 1)
elif isinstance(item_value, dict):
nested_message = getattr(parent_message, item_key).add()
for nested_key, nested_value in item_value.items():
handle_item(nested_key, nested_value, nested_message, level + 1)
else:
set_attr(item_key, item_value, parent_message)

handle_item(key, value, output_message, nesting_level)

[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 Parquet (application/x-parquet):

engine: an optional param for type of engine used to parse the parquet data,
values allowed are [auto, fastparquet, pyarrow]. If 'auto', then the behavior
is to try 'pyarrow', falling back to 'fastparquet' if ArrowNotImplementedError
is raised.
The rest of the parameters are passed unchanged to pd.DataFrame.to_parquet()
for further customizations. For additional information please see:
https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_parquet.html

For CSV (text/csv):
For parameters and general info, please see:
https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_csv.html

For JSON (application/json):
orient: indication of JSON string format to produce.
The set of possible orients is:

'split': dict like {index -> [index], columns -> [columns], data -> [values]}
'records': list like [{column -> value}, ... , {column -> value}]
'index': dict like {index -> {column -> value}}
'columns': dict like {column -> {index -> value}}
'values': just the values array
'table': dict like {'schema': {schema}, 'data': {data}}
lines: if orient is records write out line-delimited json format.
For additional parameters and general info, please see:
https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.to_json.html

For GeoJSON (application/geo+json or application/vnd.geo+json):
For parameters and general info, please see:
https://geopandas.org/docs/reference/api/geopandas.GeoDataFrame.to_json.html#geopandas.GeoDataFrame.to_json

For Protobuf (application/protobuf or application/x-protobuf):
For parameters and general info, please see:
https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_records.html

: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)
ret = None
if content_type == "application/x-parquet":
ret = GeoPandasEncoder._encode_parquet_blob(
data, engine=kwargs.get("engine", "auto"), **kwargs
)

elif content_type == "application/json":
if schema:
orient = kwargs.get("orient", "records")
lines = kwargs.get("lines", True)
ret = schema.encode_blob(
json.loads(data.to_json(orient=orient, lines=lines, **kwargs))
)
else:
ret = GeoPandasEncoder._encode_json_blob(data, **kwargs)

elif content_type == "text/csv":
ret = GeoPandasEncoder._encode_csv_blob(data, **kwargs)

elif content_type in ("application/geo+json", "application/vnd.geo+json"):
ret = (
schema.encode_blob(geojson.loads(data.to_json(**kwargs)))
if schema
else GeoPandasEncoder._encode_geojson_blob(data, **kwargs)
)

elif content_type == "application/x-protobuf":
if schema:
message_class = schema.parser.get_partition_class()
data_message = message_class()
for i in range(len(data)):
record = data.iloc[i].to_dict()
for key, value in record.items():
GeoPandasEncoder._convert_to_message(key, value, data_message)
ret = schema.encode_blob(data_message)
else:
raise ValueError(f"Schema is required to encode content type {content_type}")

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

[docs]
class GeoPandasDecoder(Decoder):
"""
Implementation of a :class:Decoder to work with pd.DataFrame and gpd.GeoDataFrame.
"""

def init(
self, including_default_value_fields: bool = True, preserving_proto_field_name: bool = True
):
"""
Initialize the decoder.

:param including_default_value_fields: if True, singular primitive fields,
repeated fields, and map fields will always be included. If False,
only include non-empty fields. Singular message fields and oneof fields are not
affected by this option. See MessageToDict.
:param preserving_proto_field_name: if True, use the original proto field
names as defined in the .proto file. If False, convert the field names
to lowerCamelCase. See MessageToDict.
"""
self._including_default_value_fields = including_default_value_fields
self._preserving_proto_field_name = preserving_proto_field_name

@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": (pd.DataFrame, gpd.GeoDataFrame), "application/x-protobuf": (pd.DataFrame, gpd.GeoDataFrame), "application/x-parquet": pd.DataFrame, "text/csv": pd.DataFrame, "application/json": pd.DataFrame, "application/vnd.geo+json": gpd.GeoDataFrame, "application/geo+json": gpd.GeoDataFrame,}

def _decode_protobuf_blob(
self,
blob: bytes,
parser: ProtobufParser,
record_path: Optional[str] = None,
record_prefix: bool = False,
max_level: Optional[int] = None,
geometry_col: Optional[str] = None,
geometry_crs: Optional[Any] = None,
**kwargs,
) -> Union[pd.DataFrame, gpd.GeoDataFrame]:
message = parser.parse_message(blob)
if not isinstance(message, Message):
raise ValueError("The decoded blob does not contain a single, valid Protobuf message")

Path is selected by the user. If None or "", the whole message is selected

record_field = parser.access_field(message, record_path)

If the field is not repeated, we consider it a collection with one element

if isinstance(record_field, Message):
record_field = [record_field]

def convert_geometry(kv: Tuple[str, Any]) -> Tuple[str, Any]:
k, v = kv
return k, to_geometry(v) if k == geometry_col else v

Lazy conversion of just the message fragments we need, one by one

def to_record(msg) -> dict:
if not isinstance(msg, Message):
raise ValueError(
f"Record path '{record_path}' does not contain a single"
" or repeated, valid Protobuf message"
)
it = iter(
MessageToDict(
msg,
always_print_fields_with_no_presence=self._including_default_value_fields,
preserving_proto_field_name=self._preserving_proto_field_name,
).items()
)
it = flatten_iterator(
it,
max_level=-1 if max_level is None else max_level,
prefix=record_path if record_prefix else "",
sep=".",
exclude_keys=[geometry_col] if geometry_col else [],
)
if geometry_col:
it = map(convert_geometry, it)
return dict(it)

The field that contains records must be repeatable, since from_records iterates over it

df = pd.DataFrame.from_records(map(to_record, record_field), **kwargs)

Final conversion to GeoDataFrame, if the geometry is present

return (
df
if not geometry_col
else gpd.GeoDataFrame(df, geometry=geometry_col, crs=geometry_crs)
)

@staticmethod
def _decode_parquet_blob(blob: bytes, engine: str, **kwargs) -> pd.DataFrame:
supported_engines = ["auto", "fastparquet", "pyarrow"]
if engine not in supported_engines:
raise ValueError(
f"The engine name is incorrect, available values are {supported_engines}"
)
if engine == "auto":
try:
return pd.read_parquet(io.BytesIO(blob), engine="auto", **kwargs)
except ArrowNotImplementedError:
logger.debug("Error in parsing with pyarrow. Trying fastparquet.")
return pd.read_parquet(io.BytesIO(blob), engine="fastparquet", **kwargs)
else:
return pd.read_parquet(io.BytesIO(blob), engine=engine, **kwargs)

@staticmethod
def _decode_csv_blob(blob: bytes, sep, header, names, index_col) -> pd.DataFrame:
return pd.read_csv(
io.BytesIO(blob), sep=sep, header=header, names=names, index_col=index_col
)

@staticmethod
def _decode_json_blob(blob: bytes, orient, lines, nrows) -> pd.DataFrame:
return pd.read_json(io.BytesIO(blob), orient=orient, lines=lines, nrows=nrows)

@staticmethod
def _decode_geojson_blob(blob: bytes) -> gpd.GeoDataFrame:
return gpd.read_file(io.BytesIO(blob))

[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
:return: the decoded blob, its type correspond to the type declared in
the property supported_content_types for the content type
:raises ValueError: in case the specified content type is not decodable
or the schema is mandatory for the content type but missing
:param kwargs: additional, content-type-specific parameters for the decoder:

For Protobuf (application/protobuf or application/x-protobuf):

record_path: the name of a schema field that is decoded and transformed to
DataFrame. It can reference nested fields by concatenating the field names
with .. When referencing a single Protobuf sub-message, that message is decoded
into one single dataframe row. When referencing repeated Protobuf messages, each
repeated message is decoded in its own row, resulting in multiple rows per partition.
Fields that are not Protobuf messages or repeated fields containing single values
(ints, strings, ...) are not supported because it is not possible to transform
them to a dataframe. If not specified, the whole blob is decoded as single message.
Messages are decoded, normalized (see max_level) and passed to
pd.DataFrame.from_record() together with the rest of kwargs: this turns each
field of the normalized messages into a column of the resulting dataframe.
record_prefix: if True, prefix the column names with the record_path.
If a non-empty string, that string is used as prefix. . is used as separator.
max_level: normalize each record of the decoded Protobuf message up to
the specified maximum level in depth. 0 disables normalization.
geometry_col: name of a column that contains geometries that is
converted to a geopandas GeoSeries, resulting in a GeoDataFrame
returned in place of a pandas DataFrame. For the supported formats,
please see documentation of here.geopandas_adapter.geo_utils.to_geometry.
Geometry field and sub-fields are excluded from normalization.
If not specified, pandas DataFrame is returned and geometry is not interpreted.
geometry_crs: the CRS to set in the GeoDataFrame, when applicable.
The rest of the parameters are passed unchanged to pd.DataFrame.from_record()
for further customizations. For additional information please see:
https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.from_records.html

For Parquet (application/x-parquet):

engine: an optional param for type of engine used to parse the parquet data,
values allowed are [auto, fastparquet, pyarrow]. If 'auto', then the behavior
is to try 'pyarrow', falling back to 'fastparquet' if ArrowNotImplementedError
is raised.
The rest of the parameters are passed unchanged to pd.read_parquet()
for further customizations. For additional information please see:
https://pandas.pydata.org/docs/reference/api/pandas.read_parquet.html

For CSV (text/csv):
sep: delimiter or column separator to use.
header: row number(s) to use as the column names, and the start of the data.
Default behavior is to infer the column names: if no names are passed the behavior is
identical to header=0 and column names are inferred from the first line
of the file, if column names are passed explicitly then the behavior
is identical to header=None. Explicitly pass header=0 to replace existing
names. The header can be a list of integers that specify row locations
for a multi-index on the columns e.g. [0,1,3]. Intervening rows that are not
specified are skipped (e.g. 2 in this example is skipped). Note that this parameter
ignores commented lines and empty lines if skip_blank_lines=True, so header=0
denotes the first line of data rather than the first line of the file.
names: list of column names to use. If the file contains a header row, then you
should explicitly pass header=0 to override the column names.
Duplicates in this list are not allowed.
index_col: column(s) to use as the row labels of the DataFrame, either given as
string name or column index. If a sequence of int/str is given, a MultiIndex is used.
Note: index_col=False can be used to force pandas to not use the first column
as the index, e.g. when you have a malformed file with delimiters
at the end of each line.

For JSON (application/json):
orient: indication of expected JSON string format.
The set of possible orients is:

'split': dict like {index -> [index], columns -> [columns], data -> [values]}
'records': list like [{column -> value}, ... , {column -> value}]
'index': dict like {index -> {column -> value}}
'columns': dict like {column -> {index -> value}}
'values': just the values array
'table': dict like {'schema': {schema}, 'data': {data}}
lines: set to True to read the file as a json object per line
nrows: the number of lines from the line-delimited json file to read. This can
only be passed if lines=True. If None, all the rows are returned.

For GeoJSON (application/geo+json or application/vnd.geo+json):
No additional parameters available.

: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)
ret = None
if content_type in ["application/protobuf", "application/x-protobuf"]:
if not schema:
raise ValueError(f"Schema is requested to decode content type {content_type}")
ret = self._decode_protobuf_blob(data, schema.parser, **kwargs)
elif content_type == "application/x-parquet":
ret = GeoPandasDecoder._decode_parquet_blob(
data, engine=kwargs.get("engine", "auto"), **kwargs
)
elif content_type == "text/csv":
ret = GeoPandasDecoder._decode_csv_blob(
data,
sep=kwargs.get("sep", ","),
header=kwargs.get("header", "infer"),
names=kwargs.get("names"),
index_col=kwargs.get("index_col"),
)
elif content_type == "application/json":
if schema:
schema.decode_blob(data)
ret = GeoPandasDecoder._decode_json_blob(
data,
orient=kwargs.get("orient"),
lines=kwargs.get("lines", False),
nrows=kwargs.get("nrows"),
)
elif content_type in ["application/vnd.geo+json", "application/geo+json"]:
if schema:
schema.decode_blob(data)
ret = GeoPandasDecoder._decode_geojson_blob(data)

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

[docs]
class GeoPandasContentAdapter(ContentAdapter):
"""
Specialization of the GeoPandasAdapter to map tabular-like content
from content bindings to GeoDataFrame or DataFrame.
"""

def init(self, partition_column: str):
"""Initialize the ContentAdapter."""
self._partition_column = partition_column

[docs]
def from_tabular(self, columns, data, geometry_column="geometry"):
"""
Convert the given attribute to tabular data
:param columns: column names
:param data: tabular data
:param geometry_column: geometry column string
:return: dataframe or geodataframe
"""

pdf = pd.DataFrame(data, columns=columns)

if geometry_column in pdf:

let's keep all changes isolated in this clone

df = pdf.copy()

identify column location

geometry_loc = df.columns.get_loc(geometry_column)

if not df.empty:
if "Point" in df.iloc[0, geometry_loc].class.name:

Point type

df[geometry_column] = df[geometry_column].apply(
lambda x: Point(x.longitude, x.latitude).wkt
)

elif "LineString" in df.iloc[0, geometry_loc].class.name:

LineString type

df[geometry_column] = df[geometry_column].apply(
lambda x: LineString(
[Point(pt.longitude, pt.latitude) for pt in x.point]
).wkt
)

convert geometric objects from WKT representation

df[geometry_column] = df[geometry_column].apply(wkt.loads)

return gpd.GeoDataFrame(df)

return original dataframe if there is no geometry column

return pdf

[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[pd.DataFrame, gpd.GeoDataFrame]:
"""
Adapt content form a structured representation to pandas :class:DataFrame or
geopandas :class:GeoDataFrame.

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.

: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 in a dataframe, indexed as requested
: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.
"""

These are hardcoded column and index names

We can make the configurable or derived smartly from ref names

These are exposed to the users

OBJ_PARTITION = self._partition_column
OBJ_ID = "identifier"
REF_PARTITION = "ref_partition"
REF_ID = "ref_identifier"
NO_VALUE = pd.NA
CRS = "EPSG:4326"

def convert_dtype(t: type):
"""
Convert type t to a (Geo)Pandas column dtype, recursively.

:param t: the type t
:return: the dtype to apply to the column, None if no conversion needed
:raises TypeError: in case of an unsupported type is encountered
"""
if t in (bool, int, float): # todo: optimize
return None # No dtype conversion
elif t in (str, Partition, Identifier):
return "string" # Enforce string
elif t is DecodedMessage:
return None # No dtype conversion, it is unpacked later
elif t in (
geojson.geometry.Geometry,
geojson.Point,
geojson.MultiPoint,
geojson.LineString,
geojson.MultiLineString,
geojson.Polygon,
geojson.MultiPolygon,
):
return "geometry" # Enforce GeoPandas geometry type
elif dataclasses.is dataclass(t): # including Ref and Range
return {f.name: convert_dtype(f.type) for f in dataclasses.fields(t)}
elif getattr(t, "origin", None) is Union:
tt, = getattr(t, "args")
return convert_dtype(tt) # todo: handle special case Option[int]
elif getattr(t, "origin", None) is dict:
_, tt = getattr(t, "args")
return None # No dtype conversion, we keep the column as-is, it is unpacked later
elif getattr(t, "origin", None) is list:
(tt,) = getattr(t, "args")
return None # No dtype conversion, we keep the column as-is
else:
raise TypeError(f"Unexpected unsupported type {t}")

def convert_value(t: type, x):
"""
Convert x to a nested list/dict representation recursively.

:param t: the type of the element
:param x: the element, None is accepted
:return: a single value, a list, or dictionary of converted values, or NA
:raises TypeError: in case of an unsupported type is encountered
"""
if t in (bool, int, float, str, Partition, Identifier, DecodedMessage):
return x if x is not None else NO VALUE
elif t in (
geojson.geometry.Geometry,
geojson.Point,
geojson.MultiPoint,
geojson.LineString,
geojson.MultiLineString,
geojson.Polygon,
geojson.MultiPolygon,
):
return shape(x) if x is not None else NO_VALUE
elif dataclasses.is_dataclass(t): # including Ref and Range
return {f.name: convert_value(f.type, getattr(x, f.name)) if x is not None else convert_value(f.type, None) for f in dataclasses.fields(t)}
elif getattr(t, "origin", None) is Union:
tt, = getattr(t, "args")
return convert_value(tt, x) if x is not None else convert_value(tt, None)
elif getattr(t, "origin", None) is dict:
_, tt = getattr(t, "args")
return (
{k: convert_value(tt, v) for k, v in x.items()} if x is not None else NO_VALUE
)
elif getattr(t, "origin", None) is list:
(tt,) = getattr(t, "args")
return [convert_value(tt, v) for v in x] if x else NO_VALUE
else:
raise TypeError(f"Unexpected unsupported type {t}")

On-the-fly validation and conversion of data

def validate_and_convert(x: object) -> dict:
x = ContentAdapter._validate_object(fields, x)
result = {f.name: convert_value(f.type, getattr(x, f.name)) for f in dataclasses.fields(fields) if f.name not in index_attrs # we avoid creating columns for indexing ...}
result = dict(flatten_iterator(iter(result.items())))

... because we need to create them manually calling the appropriate functions

if index_partition:
result["__partition"] = ContentAdapter._extract_partition(index_partition, x)
if index_id:
result["__id"] = ContentAdapter._extract_identifier(index_id, x)
if index_ref:
result["__refs"] = ContentAdapter._extract_refs(index_ref, x)
return result

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)

Names of attributes to use as-is for indexing, when present

index_partition_attr = index_partition if isinstance(index_partition, str) else None
index_id_attr = index_id if isinstance(index_id, str) else None
index_ref_attr = index_ref if isinstance(index_ref, str) else None
index_attrs = [
c for c in [index_partition_attr, index_id_attr, index_ref_attr] if c is not None
]

There is no special single element representation: return a dataframe

indexed as it is indexed in case of multiple objects

if single_element:

we may verify that data contains one element only

pass

Creation of main object DF with temporary columns to support indexing

objects_df = pd.DataFrame.from_records(map(validate_and_convert, data))

Conversion/enforcing of column types

column_dtypes = {f.name: convert_dtype(f.type) for f in dataclasses.fields(fields) if f.name not in index_attrs}
column_dtypes = {k: v for k, v in flatten_iterator(column_dtypes.items()) if k in objects_df.columns and v is not None}
if column_dtypes:
objects_df = objects_df.astype(column_dtypes)

Conversion to GeoDataFrame

geo_columns = [x for x in objects_df.columns if objects_df[x].dtype == "geometry"]
if geo_columns: # we make the first geometry the main one
objects_df = gpd.GeoDataFrame(objects_df, geometry=geo_columns[0], crs=CRS)

Indexing

if index_ref and not objects_df.empty:
assert "__refs" in objects_df.columns
ref_partition_column = REF_PARTITION
ref_id_column = REF_ID

Flatten elements with multiple references

exploded_df = objects_df.explode("__refs", ignore_index=True)

Remove elements with no references

exploded_df.dropna(subset=["__refs"], inplace=True)

Create the indexing columns

exploded_df[ref_partition_column] = exploded_df["__refs"].map(lambda r: r.partition)
exploded_df[ref_id_column] = exploded_df["__refs"].map(lambda r: r.identifier)
exploded_df.drop(columns=["__refs"], inplace=True)

Index and return

exploded_df.set_index([ref_partition_column, ref_id_column], inplace=True)
assert "__refs" not in exploded_df.columns
assert "__partition" not in exploded_df.columns
assert "__id" not in exploded_df.columns
return exploded_df
elif (index_partition or index_id) and not objects_df.empty:
new_index_columns = []
if index_partition:
assert "__partition" in objects_df.columns

Create indexing column for partition

new_partition_column = index_partition_attr or OBJ_PARTITION
objects_df.rename(columns={"__partition": new_partition_column}, inplace=True)
new_index_columns.append(new_partition_column)
if index_id:
assert "__id" in objects_df.columns

Create indexing column for identifier

new_identifier_column = index_id_attr or OBJ_ID
objects_df.rename(columns={"__id": new_identifier_column}, inplace=True)
new_index_columns.append(new_identifier_column)

Index and return

assert new_index_columns
objects_df.set_index(new_index_columns, inplace=True)
assert "__partition" not in objects_df.columns
assert "__id" not in objects_df.columns
return objects_df
else:

No indexing requested, of the dataframe is empty

assert "__refs" not in objects_df.columns
assert "__partition" not in objects_df.columns
assert "__id" not in objects_df.columns
return objects_df

[docs]
class GeoPandasAdapter(Adapter):
"""
This adapter transform data from and to pd.DataFrame and gpd.DataFrame,
when geometry information such as longitude and latitude is involved.

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:GeoPandasDecoder
and :class:GeoPandasEncoder.

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,
partition_column: str = "partition_id",
timestamp_column: str = "partition_timestamp",
including_default_value_fields: bool = True,
preserving_proto_field_name: bool = True,
):
"""Initialize the adapter, creating its components

:param partition_column: name of the partition column to add when reading
data to DataFrame or GeoDataFrame to distinguish content from
different partitions. Also used when writing data, to split content
into different partitions and encode them in different blobs.
:param timestamp_column: name of the timestamp column to add when reading
data to DataFrame or GeoDataFrame to save the Kafka message timestamp.
:param including_default_value_fields: when decoding, if True, singular primitive fields,
repeated fields, and map fields will always be included. If False,
only include non-empty fields. Singular message fields and oneof fields are not
affected by this option. See MessageToDict.
:param preserving_proto_field_name: when decoding, if True, use the original proto field
names as defined in the .proto file. If False, convert the field names
to lowerCamelCase. See MessageToDict.
"""
self._encoder = GeoPandasEncoder()
self._decoder = GeoPandasDecoder(
including_default_value_fields, preserving_proto_field_name
)
self._content_adapter = GeoPandasContentAdapter(partition_column)
self._partition_column = partition_column
self._timestamp_column = timestamp_column
self._including_default_value_fields = including_default_value_fields
self._preserving_proto_field_name = preserving_proto_field_name

@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
) -> pd.DataFrame:
"""
Adapt versioned partition metadata to a :class:pd.DataFrame.

:param partitions: sequence of partition metadata from a versioned layer
:param kwargs: unused
:return: partition metadata as :class:pd.DataFrame
"""
return self._concat_dataframes(
[
"id",
"data_handle",
"checksum",
"data_size",
"compressed_data_size",
"crc",
"version",
],
partitions,
lambda p: {"id": p.id, "data_handle": p.data_handle, "checksum": p.checksum, "data_size": p.data_size, "compressed_data_size": p.compressed_data_size, "crc": p.crc, "version": p.version,},
)

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

: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:GeoPandasDecoder
:return: partition data as :class:pd.DataFrame or :class:gpd.GeoDataFrame
"""
return self._convert_blobs_to_dataframe(partitions_data, content_type, schema, **kwargs)

[docs]
def to_versioned_metadata(
self,
layer: "VersionedLayer",
partitions_update: Optional[pd.DataFrame],
partitions_delete: Optional[pd.Series],
**kwargs,
) -> Tuple[Iterator[VersionedPartition], Iterator[Union[str, int]]]: # type: ignore
"""Adapt :class:pd.DataFrame of metadata and :class:pd.Series of keys
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 :class:pd.DataFrame of partition metadata to update, if any
:param partitions_delete: the :class:pd.Series 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
"""

TODO: this is almost a complete duplicate of volatile layer, improve

TODO: handle pd.NA, np.nan and similar

def to_partition(p) -> VersionedPartition:
assert (
not partitions_update.empty if partitions_update is not None else partitions_update
) # TODO: try to add empty columns to avoid ifs
return VersionedPartition(
data_handle=str(p.data_handle),
layer=layer,
id=str(p.id)
if partitions_update is not None and "id" in partitions_update
else None,
checksum=str(p.checksum)
if partitions_update is not None and "checksum" in partitions_update
else None,
data_size=int(p.data_size)
if partitions_update is not None and "data_size" in partitions_update
else None,
compressed_data_size=int(p.compressed_data_size)
if p.compressed_data_size
and partitions_update is not None
and "compressed_data_size" in partitions_update
else None,
crc=str(p.crc)
if partitions_update is not None and "crc" in partitions_update
else None,
)

def to_partition_id(pid) -> str:
return str(pid)

update_it: Iterator[VersionedPartition] = (
map(to_partition, partitions_update.itertuples(index=False))
if (partitions_update is not None) and not (partitions_update.empty)
else iter([]) # type: ignore
)
delete_it = (
map(to_partition_id, partitions_delete)
if (partitions_delete is not None) and not (partitions_delete.empty)
else iter([]) # type: ignore
)
return update_it, delete_it

[docs]
def to_versioned_data(
self,
layer: "VersionedLayer",
data: "pd.DataFrame",
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: data as :class:pd.DataFrame or :class:gpd.GeoDataFrame
: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:GeoPandasEncoder
:return: sequence of partition id and data for the volatile layer
"""
return self._convert_dataframe_to_blobs(data, content_type, schema, **kwargs)

[docs]
def from_volatile_metadata(
self, partitions: Iterator[VolatilePartition], **kwargs
) -> pd.DataFrame:
"""Adapt volatile partition metadata to the target format.

:param partitions: sequence of partition metadata from a volatile layer
:param kwargs: unused
:return: partition metadata as :class:pd.DataFrame
"""
return self._concat_dataframes(
["id", "data_handle", "checksum", "data_size", "compressed_data_size", "crc"],
partitions,
lambda p: {"id": p.id, "data_handle": p.data_handle, "checksum": p.checksum, "data_size": p.data_size, "compressed_data_size": p.compressed_data_size, "crc": p.crc,},
)

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

: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:GeoPandasDecoder
:return: partition data as :class:pd.DataFrame or :class:gpd.GeoDataFrame
"""
return self._convert_blobs_to_dataframe(partitions_data, content_type, schema, **kwargs)

[docs]
def to_volatile_metadata(
self,
layer: "VolatileLayer",
partitions_update: Optional[pd.DataFrame],
partitions_delete: Optional[pd.Series],
**kwargs,
) -> Tuple[Iterator[VolatilePartition], Iterator[Union[str, int]]]:
"""Adapt :class:pd.DataFrame of metadata and :class:pd.Series of keys
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 :class:pd.DataFrame of partition metadata to update, if any
:param partitions_delete: the :class:pd.Series 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
"""

TODO: handle pd.NA, np.nan and similar

def to_partition(p) -> VolatilePartition:
assert (
not partitions_update.empty if partitions_update is not None else partitions_update
) # TODO: try to add empty columns to avoid ifs
return VolatilePartition(
data_handle=str(p.data_handle),
layer=layer,
id=str(p.id)
if partitions_update is not None and "id" in partitions_update
else None,
checksum=str(p.checksum)
if partitions_update is not None and "checksum" in partitions_update
else None,
data_size=int(p.data_size)
if partitions_update is not None and "data_size" in partitions_update
else None,
compressed_data_size=int(p.compressed_data_size)
if p.compressed_data_size
and partitions_update is not None
and "compressed_data_size" in partitions_update
else None,
crc=str(p.crc)
if partitions_update is not None and "crc" in partitions_update
else None,
)

def to_partition_id(pid) -> str:
return str(pid)

update_it: Iterator[VolatilePartition] = (
map(to_partition, partitions_update.itertuples(index=False))
if (partitions_update is not None) and not (partitions_update.empty)
else iter([]) # type: ignore
)
delete_it = (
map(to_partition_id, partitions_delete)
if (partitions_delete is not None) and not (partitions_delete.empty)
else iter([]) # type: ignore
)
return update_it, delete_it

[docs]
def to_volatile_data(
self,
layer: "VolatileLayer",
data: "pd.DataFrame",
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.

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

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

return self._concat_dataframes(
[
"id",
"data_handle",
"data_size",
"data",
"checksum",
"crc",
"timestamp",
"kafka_partition",
"kafka_offset",
],
partitions,
lambda p: {"id": p.id, "data_handle": p.data_handle, "data_size": p.data_size, "data": p.data, "checksum": p.checksum, "crc": p.crc, "timestamp": pd.Timestamp(p.timestamp, unit="ms") if p.timestamp else pd.NA, "kafka_partition": p.kafka_partition, "kafka_offset": p.kafka_offset,},
)

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

: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:GeoPandasDecoder
:return: stream message data as :class:pd.DataFrame or :class:gpd.GeoDataFrame
"""
return self._convert_blobs_to_dataframe(partitions_data, content_type, schema, **kwargs)

[docs]
def to_stream_metadata(
self, layer: "StreamLayer", partitions: pd.DataFrame, **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: the :class:pd.DataFrame of partition metadata to append
:param kwargs: unused
:yield: the :class:StreamPartition that are adapted
"""
for p in partitions.itertuples(index=False):

Kafka partition and offset are not copied, as these are determined

by kafka when writing, we can't specify arbitrarily values for them.

yield StreamPartition(
layer=layer,
id=_extract(p, "id", str),
data_handle=_extract(p, "data_handle", str),
data_size=_extract(p, "data_size", int),
data=_extract(p, "data", bytes),
checksum=_extract(p, "checksum", str),
crc=_extract(p, "crc", str),
timestamp=_extract(
p,
"timestamp",
lambda t: int(t.timestamp() * 1000) if isinstance(t, pd.Timestamp) else t,
),
)

[docs]
def to_stream_data(
self,
layer: "StreamLayer",
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: adapter-specific, the data 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
:yield: partition id, data and timestamp for the stream layer
:raises ValueError: in case required columns are missing
"""

code comes from _convert_dataframe_to_blobs

with added support for kafka timestamp

if self._partition_column not in data:
raise ValueError(
f"DataFrame does not contain column {self._partition_column} "
"needed for partitioning"
)

This should be improved asking pandas to group-by partition column

and then process each group for encoding in the next for loop

data = self.convert_to_dict(data)
partitions_ids = data[self._partition_column].dropna().unique()

for partition_id in partitions_ids:
partition_data_complete = data[data[self._partition_column] == partition_id]

if self._timestamp_column in partition_data_complete:
timestamp_column = (
partition_data_complete[self._timestamp_column]
.astype("datetime64[ms]")
.dropna()
)
timestamp_value = timestamp_column.iloc[0] if not timestamp_column.empty else None
ts = int(timestamp_value.timestamp() * 1000) if timestamp_value else None
partition_data = partition_data_complete.drop(
columns=[self._partition_column, self._timestamp_column]
)
else:
partition_data = partition_data_complete.drop(columns=[self._partition_column])
ts = None

encoded = self._encoder.encode_blob(partition_data, content_type, schema, **kwargs)
yield str(partition_id), encoded, (ts or timestamp)

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

:param partitions: sequence of partition metadata from an index layer
:param kwargs: unused
:return: partition metadata as :class:pd.DataFrame
"""
return self._concat_dataframes(
["id", "data_handle", "checksum", "data_size", "crc"],
partitions,
lambda p: {"id": p.data_handle, "data_handle": p.data_handle, "checksum": p.checksum, "data_size": p.data_size, "crc": p.crc,},
)

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

: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:GeoPandasDecoder
:return: partition data as :class:pd.DataFrame or :class:gpd.GeoDataFrame
"""
return self._convert_blobs_to_dataframe(partitions_data, content_type, schema, **kwargs)

def _convert_blobs_to_dataframe(
self,
partitions_data: Iterator[Tuple[Partition, bytes]],
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> pd.DataFrame:
"""
Convert decoded blobs to DataFrame based on layer schema.

:param partitions_data: sequence of partition metadata and data from a 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 for the adapter
:return: partition data as :class:pd.DataFrame
"""

TODO make sure specialized metadata is decoded,

e.g. timestamps and similar fields in IndexPartition

def decode_df(partition_data):
partition, data = partition_data
df = self.decoder.decode_blob(data, content_type, schema, **kwargs)
assert isinstance(df, pd.DataFrame)
if isinstance(partition, StreamPartition):
ts = pd.Timestamp(partition.timestamp, unit="ms") if partition.timestamp else pd.NA
df.insert(0, self._timestamp_column, ts)
df.insert(0, self._partition_column, partition.id)
return df

decoded_df = map(decode_df, partitions_data)

TODO: improve it by creating the final DF all at once and not appending iteratively

output_df = pd.DataFrame()
for decoded_df in decoded_df:
output_df = pd.concat([output_df, decoded_df], ignore_index=True)
return output_df

[docs]
@staticmethod
def convert_to_dict(df):
"""Converts the columns to dict"""
for column_name in df.columns:
if "." in column_name:
prefix = column_name.split(".")[0]
cols = [col for col in df.columns if col.startswith(prefix)]
df1 = df.drop(columns=cols)
in_dict = pd.Series(df[cols].to_dict("index"))
in_dict = in_dict.apply(
lambda x: {k.replace(prefix + ".", ""): v for k, v in x.items()}
)
df = df1.assign(**{prefix: in_dict})
continue
for i in range(len(df[column_name])):
value = df.at[i, column_name]
if not isinstance(value, list):
continue
if not isinstance(value[0], dict):
df.at[i, column_name] = value
continue
keys = [k for k in value[0].keys() if "." in k]
nested_key = keys[0].split(".")[0] if keys else None
if nested_key:
df.at[i, column_name] = [
{nested_key: { k.split(".")[1]: v for k, v in value[0].items() if k.startswith(nested_key + ".") }, **{k: v for k, v in value[0].items() if "." not in k},}
]
else:
df.at[i, column_name] = value
return df

def _convert_dataframe_to_blobs(
self,
data: "pd.DataFrame",
content_type: str,
schema: Optional[Schema],
**kwargs,
) -> Iterator[Tuple[str, bytes]]:
if self._partition_column not in data:
raise ValueError(
f"DataFrame does not contain column {self._partition_column} "
"needed for partitioning"
)

def check_list(col):
return all(isinstance(i, (list, dict)) for i in col)

data = self.convert_to_dict(data)

agg_dict =
for col in data.columns:
if check_list(data[col]):
agg_dict[col] = pd.Series.tolist
else:
agg_dict[col] = "first"

data_processed = data.groupby(self._partition_column).agg(agg_dict)

partition_ids = data_processed[self._partition_column].unique()
for partition_id in partition_ids:
partition_data = data_processed[
data_processed[self._partition_column] == partition_id
].drop(columns=self._partition_column)
encoded = self._encoder.encode_blob(partition_data, content_type, schema, **kwargs)
yield str(partition_id), encoded

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

:param data: data in the form of DataFrame
: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:GeoPandasEncoder
:return: data encoded for an index layer
:raises ValueError: in case the content type is not supported by the adapter # noqa
"""
data = self.convert_to_dict(data)
ret: bytes = self.encoder.encode_blob(data, content_type, schema, **kwargs)
return ret

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

:param feature_ids: sequence of feature identifiers
:param kwargs: additional parameters are passed unchanged to
pd.Series(). For additional information please see:
https://pandas.pydata.org/docs/reference/api/pandas.Series.html
:return: a Series with the feature identifiers
"""
return pd.Series(data=feature_ids, **kwargs)

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

Values are converted to str. NA values discarded.

:param data: a Series containing feature identifiers
:param kwargs: unused
:return: sequence of feature identifiers
"""

def tofeature_id(idx_value) -> str: , value = idx_value
return str(value)

return map(to_feature_id, data.dropna().items())

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

:param features: sequence of geographic features
:param kwargs: additional parameters are passed unchanged to
gpd.GeoDataFrame.from_features(). For additional information please see:
https://geopandas.org/docs/reference/api/geopandas.GeoDataFrame.from_features.html
:return: a new gpd.GeoDataFrame containing the features
"""

Object identifiers are lost due to a limitation of GeoPandas,

see https://github.com/geopandas/geopandas/issues/1988.

Therefore we construct the index and apply it manually.

When the bug in GeoPandas is fixed, we can upgrade to the new version by requiring

at least the fixed version in requirements.txt and get rid of this logic

ids = []

def get_feature_and_save_id(f):
ids.append(f.get("id"))
return f

features_iter = map(get_feature_and_save_id, features)
gdf = gpd.GeoDataFrame.from_features(features=features_iter, **kwargs)
gdf = gdf.set_index(pd.Index(ids))
return gdf

[docs]
def to_geo_features(self, data: gpd.GeoDataFrame, **kwargs) -> Iterator[Feature]:
"""Adapt data in a GeoDataFrame to a sequence of geographic features.

:param data: the gpd.GeoDataFrame to adapt
:param kwargs: additional parameters are passed unchanged to
gpd.GeoDataFrame.iterfeatures(). For additional information please see:
https://geopandas.org/docs/reference/api/geopandas.GeoDataFrame.iterfeatures.html
:return: sequence of geographic features from the GeoDataFrame
"""

def to_feature(geof):
return Feature(
id=geof.get("id"),
geometry=geof.get("geometry"),
properties=geof.get("properties"),
)

return map(to_feature, data.iterfeatures(**kwargs))

def _concat_dataframes(
self, columns: list, partitions: Iterator[IndexPartition], map_data: typing.Callable
) -> pd.DataFrame:
"""
Extracts data from each partition in :param:partitions, converts them into a dataframe
using the function :param:map_data and dumps all into a resultant dataframe with given
:param:columns.

:param columns: columns needed in the resultant dataframe
:param partitions: Iterator of partitions to be mapped/parsed
:param map_data: lambda function to parse data in :param:partitions
:return: resultant dataframe with all the mapped/parsed data
"""

partitions_df = pd.DataFrame(columns=columns)
for p in partitions:
partitions_df = pd.concat(
[partitions_df, pd.DataFrame([map_data(p)])], ignore_index=True, axis=0
)
return partitions_df