Copyright (C) 2020-2023 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 GeoJson Parser module
"""

import io
from pathlib import Path
from typing import Union

import geojson
from geojson import FeatureCollection
from google.protobuf.message import Message
from here.platform.schema.json_parser import JsonParser

[docs]
class GeoJsonParser(JsonParser):
"""A class for parsing GeoJson data of a catalog layer partition."""

def init(self, schema_pkg_file: Union[str, Path, io.BytesIO]):
"""
Initialize a GeoJsonParser instance.

:param schema_pkg_file: the schema package zip-file as a str, Path or file-like object
"""
super().init(schema_pkg_file)

[docs]
def parse_message(self, message: bytes) -> Union[Message, dict]:
"""
Parse a message and returns the corresponding partition class instance.

:param message: message received as blob
:raises ValueError: if the decoded data doesn't comply with geojson standards.
:return: parsed message
"""
decoded = geojson.loads(message)
self.validate_json_schema(decoded)
if not isinstance(decoded, FeatureCollection):
raise ValueError("Blob does not contain a valid GeoJSON FeatureCollection")
assert isinstance(decoded, dict)
return decoded

[docs]
def serialize_bytes(self, data: Union[Message, dict]) -> bytes:
"""
Parse a GeoJson message and returns the bytes.

:param data: GeoJson message
:return: bytes
"""
assert isinstance(data, dict)
self.validate_json_schema(data)
message_bytes = geojson.dumps(data).encode("utf-8")
assert isinstance(message_bytes, bytes)
return message_bytes