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 Parser module
"""

import io
import json
import logging
import zipfile
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Optional, Tuple, Union

from google.protobuf.message import Message

logger = logging.getLogger(name)

[docs]
class Parser(ABC):
"""Base class for parsing data of a catalog layer partition."""

LAYER_MANIFEST = "layer.manifest.json"
LAYER_DESCRIPTORS = "layer.fds"

[docs]
@staticmethod
def extract_layer_info(
input_zip: Union[str, Path, io.BytesIO]
) -> Tuple[Optional[dict], Optional[bytes]]:
"""
Extract layer info from an input zipfile.

:param input_zip: input Protobuf schema zipfile
:returns: content of the layer manifest and descriptor set
"""
input_zipfile = zipfile.ZipFile(input_zip)
layer_manifest = None
layer_descriptor_set = None
for name in input_zipfile.namelist():
if name.endswith(Parser.LAYER_MANIFEST):
layer_manifest = json.loads(input_zipfile.read(name).decode("utf-8"))
elif name.endswith(Parser.LAYER_DESCRIPTORS):
layer_descriptor_set = input_zipfile.read(name)
return layer_manifest, layer_descriptor_set

[docs]
@abstractmethod
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
:return: parsed message
"""

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

:param data: message
:return: bytes
"""