Source code for here.content.hmc2.base_attributes

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.

"""
Common definition and helper functions for HMC attributes.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Iterator, List, Mapping, Optional

from google.protobuf.message import Message
from here.content.base import ContentIndexer, ContentPartition, SingleLayerBinding
from here.content.content import Content as ParentContent
from here.content.utils.proto import decode_message, parse_ref
from here.platform.adapter import Adapter, DecodedMessage, Range, Ref, make_range

[docs]
@dataclass
class SegmentRef:
"""A reference to a specific direction and range of segment"""

ref: Ref
inverted: bool
offset: Range

def hash(self):
"""Return hash(self)."""
return hash(
self.ref.partition

self.ref.identifier
str(self.inverted)
str(self.offset.start)
str(self.offset.end)
)

@dataclass
class _SegmentAnchor:
"""
An anchor to one or more segments, each with its range and relative direction.

The class is private as it is used only for parsing.
The concept is flattened out to simplify user experience.
"""

segments: List[SegmentRef]
orientation: str

[docs]
@dataclass
class Attribute:
"""An attribute of a specific direction and sequence of segments where it applies"""

segments: List[SegmentRef]
forward: bool
backward: bool
attribute: DecodedMessage

indexer = ContentIndexer(
refs={"first_segment": lambda a: a.segments[0].ref, # type: ignore "segment": lambda a: [s.ref for s in a.segments], # type: ignore}
)

[docs]
class BaseAttributesBinding(SingleLayerBinding, ABC):
"""
Bindings for HCM road attributes.
"""

@classmethod
def parse_segment_anchor(cls, msg) -> SegmentAnchor:
"""
Parse a protobuf message that contains a segment anchor.

:param msg: the protobuf message
:return: the parsed segment anchor
"""
start_offset = (
msg.first_segment_start_offset.value
if msg.HasField("first_segment_start_offset")
else None
)
end_offset = (
msg.last_segment_end_offset.value if msg.HasField("last_segment_end_offset") else None
)
n_segments = len(msg.oriented_segment_ref)
return _SegmentAnchor(
segments=[
SegmentRef(
ref=parse_ref(ref.segment_ref),
inverted=ref.inverted,
offset=make_range(
start_offset if n == 0 else None,
end_offset if n == (n_segments - 1) else None,
), # todo: check if there's the need to invert in case the segment is inverted
)
for n, ref in enumerate(msg.oriented_segment_ref)
],
orientation=msg.attribute_orientation,
)

@classmethod
def _parse_generic_attributes(
cls, segment_anchors: List[_SegmentAnchor], msg
) -> Iterator[Attribute]:
"""
Parse protobuf messages that contain generic attributes,
each with references to applicable segments, out of a pool of segment anchors.

:param segment_anchors: the pool of segment anchors, the function
looks up each anchor its index in the list
:param msg: repeated protobuf message containing the attributes
:yield: parsed attributes
"""

for attribute_msg in msg:
a_value = decode_message(attribute_msg, ignore=["segment_anchor_index"])
for sa_index in attribute_msg.segment_anchor_index:
sa = segment_anchors[sa_index]

in case the anchor is malformed and doesn't contain any reference

if not sa.segments:
continue
yield Attribute(
segments=sa.segments,
forward=sa.orientation in (1, 2),
backward=sa.orientation in (1, 3),
attribute=a_value,
)

[docs]
@classmethod
def parse_and_index(cls, partition_id: str, msg: Message) -> ContentPartition:
"""
Parse and index decoded message.

:param partition_id: identifier of the partition
:param msg: decoded content from the layer, in the form of a Protobuf message
:return: the content of a partition parsed and indexed in its own data structure
"""
assert hasattr(msg, "segment_anchor")

parse segment anchors

segment_anchors = [
cls._parse_segment_anchor(sa) for sa in msg.segment_anchor # type: ignore
]

segment_attributes = {
attribute_type: list(
cls._parse_generic_attributes(segment_anchors, getattr(msg, attribute_type))
)
for attribute_type in cls.object_types()

the following is to avoid parsing attributes that are empty

if hasattr(msg, attribute_type) and getattr(msg, attribute_type)
}

return ContentPartition(cls, **segment_attributes)

[docs]
class BaseAttributesMultiLayerBinding(BaseAttributesBinding, ABC):
"""
Base Attributes Multilayer Binding Class
"""

def init(self, content: ParentContent, layer_id: str, adapter: Optional[Adapter] = None):
"""
Instantiate an class BaseAttributes MultiLayer binding.

:param content: Content object that references a catalog at a fixed version
:param layer_id: ID of the only layer that contains the content
:param adapter: the Adapter to transform data between different representations,
None to use the adapter defined in content.
"""
super().init(content, layer_id, adapter)

[docs]
@classmethod
@abstractmethod
def associated_layers(cls) -> Dict[str, SingleLayerBinding]:
"""
Return associating layers to the base layer defined in schema hrn.

:return: dict of key as single layer and value as SingleLayerBinding Object.
"""

@classmethod
@abstractmethod
def _internal_object_types(cls) -> Mapping[str, type]:
"""
Required for primary layers object type which need to parsed_and_indexed prior to
associating data with dependent layers and should not be visible to users.
Return private object types which should not be visible to users.

:return: dict of the name and types of objects supported by this binding
"""