Copyright (C) 2021-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.

"""
Utility to handle data encoded in protobuf format
"""
from typing import Any, Iterable, List

from geojson import LineString, MultiPolygon, Point
from google.protobuf.json_format import MessageToDict
from google.protobuf.message import Message
from here.platform.adapter import DecodedMessage, Ref, make_ref

Useful functions for HMC

[docs]
def parse_ref(msg) -> Ref:
"""Parse a HMC reference."""
return make_ref(msg.partition_name, msg.identifier)

[docs]
def parse_point(msg) -> Point:
"""Parse an HMC point with optional elevation."""
if hasattr(msg, "elevation") and msg.elevation:
return Point([msg.longitude, msg.latitude, msg.elevation])
else:
return Point([msg.longitude, msg.latitude])

[docs]
def parse_linestring(msg) -> LineString:
"""Parse an HMC line string with optional elevation."""
if any((hasattr(p, "elevation") and p.elevation for p in msg.point)):
return LineString([(p.longitude, p.latitude, p.elevation) for p in msg.point])
else:
return LineString([(p.longitude, p.latitude) for p in msg.point])

[docs]
def parse_multi_polygon(msg) -> MultiPolygon:
"""Parse an HMC multi polygon with optional elevation."""
multi_polygon = []

for polygon in msg.multi_polygon.polygon:
poly_rings = [parse_linestring(polygon.exterior_ring)]

for ring in polygon.interior_ring:
poly_rings.append(parse_linestring(ring))
multi_polygon.append(poly_rings)

return MultiPolygon(multi_polygon)

Useful generic functions

[docs]
def decode_message(msg: Message, ignore: List[str] = []) -> DecodedMessage:
"""
Decode a protobuf message to a dictionary, recursively.

:param msg: the protobuf message to convert
:param ignore: the name of fields to ignore
:return: the resulting dictionary
"""
msg_dict = MessageToDict(
msg, preserving_proto_field_name=True, always_print_fields_with_no_presence=True
)

This is not efficient in case ignore is set, because the

ignored fields are converted and then discarded.

We can modify the object on the fly because it is local.

for k in ignore:
if k in msg_dict:
del msg_dict[k]
return DecodedMessage(msg_dict)

[docs]
def decode_enum(msg: Message, field: str) -> str:
"""
Decode an enum field of a message to string.

:param msg: the protobuf message containing the field
:param field: the name of fields
:return: the resulting enum converted to string
:raises KeyError: in case the message doesn't contain the expected field
:raises ValueError: in case the value can not be mapped to an enum value
"""
for field_desc, value in msg.ListFields():
if field_desc.name == field:

copied from json_format.py of protobuf

enum_value = field_desc.enum_type.values_by_number.get(value, None)
if enum_value is not None:
return enum_value.name # type: ignore
elif field_desc.file.syntax == "proto3": # type: ignore
return value # type: ignore
else:
raise ValueError(
"Enum field contains an integer value which can not mapped to an enum value."
)

When the enum type is empty, return the default value

for field_desc in msg.DESCRIPTOR.fields:
if field_desc.name == field and field_desc.enum_type is not None:
enum_value = field_desc.enum_type.values_by_number.get(0, None)
if enum_value is not None:
return enum_value.name # type: ignore

raise KeyError(f"Field {field} not present in {msg}") # noqa: E713

[docs]
def decode_enum_array(msg: Message, field: str, is_required: bool = True) -> List[str]:
"""
Decode an enum field of a message to string.

:param msg: the protobuf message containing the field
:param field: the name of fields
:param is_required: determines if the field is a required field
:return: the resulting enum values converted to a list of string
:raises KeyError: in case the field is required & message doesn't contain the expected field
:raises ValueError: in case the value can not be mapped to an enum value
:raises TypeError: in case the value of the field is not iterable
"""
enum_values = []
is_field_present = False
for field_desc, value in msg.ListFields():
if field_desc.name == field:
is_field_present = True
if not isinstance(value, Iterable):
raise TypeError(f'The value of the field "{field}" is not iterable')

copied from json_format.py of protobuf

for val in value:
enum_value = field_desc.enum_type.values_by_number.get(val, None)
if enum_value is not None:
enum_values.append(enum_value.name) # type: ignore
elif field_desc.file.syntax == "proto3": # type: ignore
enum_values.append(val) # type: ignore
else:
raise ValueError(
"Enum field contains an integer value "

• "which cannot be mapped to an enum value."
  )
  if is_required and not is_field_present:
  raise KeyError(f'Field "{field}" is required but not present in "{msg}"') # noqa: E713
  return enum_values

[docs]
def has_field(msg: Message, field: str) -> bool:
"""
Checks whether field is present in the Message
and returns boolean value instead of raising a ValueError.

:param msg: the protobuf message containing the field.
:param field: the name of field.
:returns: whether field is present in the message.
"""
try:
return msg.HasField(field)
except ValueError:
return False

[docs]
def has_field_attr(msg: Message, field: str) -> bool:
"""
Checks whether field is present in the Message
and returns boolean value instead of raising a ValueError.

:param msg: the protobuf message containing the field.
:param field: the name of field.
:returns: whether field is present in the message.
"""
if has_field(msg, field):
return True
elif hasattr(msg, field):
return True
else:
return False

[docs]
def value_or_None(msg: Message, field: str) -> Any:
"""
Returns the value of a field if the field exists.

:param msg: the protobuf message containing the field.
:param field: the name of field.
:returns: value of the field if the field exists or None otherwise.
"""
if has_field(msg, field):
return getattr(msg, field).value
else:
return None